pyrtlsdr

https://pyrtlsdr.readthedocs.io/en/latest/
Admin
pyrtlsdr is a Python library that provides a convenient interface for controlling and receiving...
Tokens:27,296
Snippets:140
Trust Score:8.3
Update:11 months ago
Show doc for...
Tokens:
Raw
### Run RTL-SDR TCP Server

Source: https://pyrtlsdr.readthedocs.io/en/latest/Overview

Example showing how to start an RtlSdrTcpServer which allows network access to a physically connected RTL-SDR device. Listens for incoming connections until manually stopped.

```python
server = RtlSdrTcpServer(hostname='192.168.1.100', port=12345)
server.run_forever()
# Will listen for clients until Ctrl-C is pressed
```

--------------------------------

### Read and Print RTL-SDR Samples

Source: https://pyrtlsdr.readthedocs.io/en/latest/Reference

A simple Python example demonstrating how to initialize an RTL-SDR device, read a specified number of samples, and print them. This requires the pyrtlsdr library and a connected RTL-SDR dongle.

```python
from rtlsdr import RtlSdr

sdr = RtlSdr()
sdr.sample_rate = 2.06e6  # Hz
sdr.center_freq = 433.92e6  # Hz
sdr.gain = 'auto'

# Read 16384 samples
samples = sdr.read_samples(16384)

print(samples)
sdr.close()

```

--------------------------------

### Asynchronous Streaming with RtlSdrAio

Source: https://pyrtlsdr.readthedocs.io/en/latest/rtlsdraio

Demonstrates how to use the RtlSdrAio class to asynchronously stream samples from an RTL-SDR device. The example shows setting up the device, streaming samples, and properly stopping the stream.

```python
import asyncio
from rtlsdr import RtlSdr

async def streaming():
    sdr = RtlSdr()

    async for samples in sdr.stream():
        # do something with samples
        # ...

    # to stop streaming:
    await sdr.stop()

    # done
    sdr.close()

loop = asyncio.get_event_loop()
loop.run_until_complete(streaming())
```

--------------------------------

### Python RTL-SDR Library Initialization and Basic Settings

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

This Python snippet demonstrates the basic setup for the RTL-SDR library, including importing necessary modules, handling Python version compatibility, and defining constants related to device settings like gain, center frequency, and sample rate. It also includes methods for interacting with the underlying librtlsdr library.

```python
#    This file is part of pyrlsdr.
#    Copyright (C) 2013 by Roger <https://github.com/roger-/pyrtlsdr>
#
#    pyrlsdr is free software: you can redistribute it and/or modify
#    it under the terms of the GNU General Public License as published by
#    the Free Software Foundation, either version 3 of the License, or
#    (at your option) any later version.

#    pyrlsdr is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
#    You should have received a copy of the GNU General Public License
#    along with pyrlsdr.  If not, see <http://www.gnu.org/licenses/>.


from __future__ import division, print_function
from ctypes import *
from .librtlsdr import (
    librtlsdr,
    p_rtlsdr_dev,
    rtlsdr_read_async_cb_t,
    tuner_bandwidth_supported,
    tuner_set_bandwidth_supported,
)
try:
    from itertools import izip
except ImportError:
    izip = zip
import sys

PY3 = sys.version_info.major >= 3
if PY3:
    basestring = str

```

--------------------------------

### Plot PSD with matplotlib

Source: https://pyrtlsdr.readthedocs.io/en/latest/Overview

Example showing how to capture samples from an RTL-SDR device and plot the power spectral density using matplotlib. Configures the device with specific sample rate, center frequency, and gain settings before reading 256K samples.

```python
from pylab import *
from rtlsdr import *

sdr = RtlSdr()

# configure device
sdr.sample_rate = 2.4e6
sdr.center_freq = 95e6
sdr.gain = 4

samples = sdr.read_samples(256*1024)
sdr.close()
```

--------------------------------

### Python: Handle Property Get

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/server

Handles incoming property get requests. It retrieves the property name from the request, validates the property, retrieves the property value from the RTL-SDR device, and sends the value back to the client.

```python
def handle_prop_get(self, rx_message):
    prop_name = rx_message.header.get('name')
    if prop_name not in API_DESCRIPTORS:
        raise CommunicationError('property %s not allowed' % (prop_name))
    rtl_sdr = self.server.rtl_sdr
    value = getattr(rtl_sdr, prop_name)
    tx_message = ServerMessage(client_message=rx_message, data=value)
    tx_message.send_message(self.request)
```

--------------------------------

### GET /api/device/gains

Source: https://pyrtlsdr.readthedocs.io/en/latest/genindex

Returns all available gain values supported by the RTLSDR device. Useful for determining valid gain settings before configuration.

```APIDOC
## GET /api/device/gains

### Description
Returns all available gain values supported by the RTLSDR device.

### Method
GET

### Endpoint
/api/device/gains

### Response
#### Success Response (200)
- **gains** (array) - Array of supported gain values

#### Response Example
{
  "gains": [0.0, 0.9, 1.4, 2.7, 3.7, 7.7, 8.7, 12.5, 14.4, 15.7, 16.6, 19.7, 20.7, 22.9, 25.4, 28.0, 29.7, 32.8, 33.8, 36.4, 37.2, 38.6, 40.2, 42.1, 43.4, 43.9, 44.5, 48.0, 49.6]
}
```

--------------------------------

### GET /api/device/bandwidth

Source: https://pyrtlsdr.readthedocs.io/en/latest/genindex

Returns the current bandwidth setting of the RTLSDR device. This helps determine the effective frequency range being captured.

```APIDOC
## GET /api/device/bandwidth

### Description
Returns the current bandwidth setting of the RTLSDR device.

### Method
GET

### Endpoint
/api/device/bandwidth

### Response
#### Success Response (200)
- **bandwidth** (number) - The bandwidth in Hz

#### Response Example
{
  "bandwidth": 1000000
}
```

--------------------------------

### GET /api/device/sample_rate

Source: https://pyrtlsdr.readthedocs.io/en/latest/genindex

Retrieves the current sample rate setting of the RTLSDR device. Sample rate determines how much bandwidth can be captured.

```APIDOC
## GET /api/device/sample_rate

### Description
Retrieves the current sample rate setting of the RTLSDR device.

### Method
GET

### Endpoint
/api/device/sample_rate

### Response
#### Success Response (200)
- **sample_rate** (number) - The sample rate in samples per second

#### Response Example
{
  "sample_rate": 2048000
}
```

--------------------------------

### RequestHandler for Client Connections

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/server

Handles individual client connections to the TCP server. The `setup` method is called when a new request is received, initializing the handler's state. This class is intended to be extended to manage communication protocols and data exchange with clients.

```python
class RequestHandler(BaseRequestHandler):

    def setup(self):
        self.finished = False

```

--------------------------------

### Handle Multiple RTL-SDR Devices

Source: https://pyrtlsdr.readthedocs.io/en/latest/Reference

This example demonstrates how to use pyrtlsdr to detect and interact with multiple RTL-SDR devices connected to the system. It retrieves a list of available devices and allows selection based on index or serial number. This is useful for parallel signal acquisition.

```python
from rtlsdr import RtlSdr

# Get a list of available devices
num_devices = RtlSdr.get_device_count()
print(f"Found {num_devices} device(s)\n")

# Get serial numbers for all devices
serials = RtlSdr.get_device_serial_addresses()
print(f"Device serials: {serials}\n")

# Example: Open the first device
if num_devices > 0:
    sdr = RtlSdr(device_index=0)
    print(f"Opened device: {sdr.device_serial}\n")
    # Configure and use the device...
    sdr.close()
else:
    print("No RTL-SDR devices found.")

```

--------------------------------

### Plot RTL-SDR Power Spectral Density (PSD)

Source: https://pyrtlsdr.readthedocs.io/en/latest/Reference

This Python code snippet shows how to read samples from an RTL-SDR device and plot their Power Spectral Density (PSD) using matplotlib. It requires the pyrtlsdr and matplotlib libraries. The example sets the sample rate, center frequency, and gain before reading and plotting.

```python
import matplotlib.pyplot as plt
from rtlsdr import RtlSdr
import numpy as np

sdr = RtlSdr()
sdr.sample_rate = 2.4e6  # Hz
sdr.center_freq = 915.25e6  # Hz
sdr.gain = 'auto'

# Read samples and calculate PSD
samples = sdr.read_samples(16384)
psd = np.fft.fftshift(np.fft.fft(samples))
freqs = np.fft.fftshift(np.fft.fftfreq(len(samples), 1/sdr.sample_rate))

plt.figure()
plt.plot(freqs, np.abs(psd)
plt.xlabel("Frequency (Hz)")
plt.ylabel("PSD")
plt.title("Power Spectral Density")
plt.show()
sdr.close()

```

--------------------------------

### Set and Get Sample Rate

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Configures the device sample rate and retrieves the effective rate computed from the crystal frequency and rsamp ratio. Validates librtlsdr return codes and closes the device on error.

```python
def set_sample_rate(self, rate):

    rate = int(rate)

    result = librtlsdr.rtlsdr_set_sample_rate(self.dev_p, rate)
    if result < 0:
        self.close()
        raise LibUSBError(result, 'Could not set sample rate to %d Hz' % (rate))

    return

def get_sample_rate(self):

    result = librtlsdr.rtlsdr_get_sample_rate(self.dev_p)
    if result < 0:
        self.close()
        raise LibUSBError(result, 'Could not get sample rate')

    # figure out actual sample rate, taken directly from librtlsdr
    reported_sample_rate = result
    rsamp_ratio = (self.CRYSTAL_FREQ * pow(2, 22)) // reported_sample_rate
    rsamp_ratio &= ~3
    real_rate = (self.CRYSTAL_FREQ * pow(2, 22)) / rsamp_ratio;

    return real_rate
```

--------------------------------

### Set and Get Tuner Bandwidth with Version Fallbacks

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Sets the tuner bandwidth using newer bandwidth query APIs when available, falling back to older set-only API, and raises IOError if unsupported. Caches the applied bandwidth for later retrieval.

```python
def set_bandwidth(self, bw):

    requested_bw = int(bw)
    bw = int(bw)
    if tuner_bandwidth_supported:
        apply_bw = c_int(1)
        applied_bw = c_uint32(bw)
        bw = c_uint32(bw)
        result = librtlsdr.rtlsdr_set_and_get_tuner_bandwidth(
            self.dev_p, bw, byref(applied_bw), apply_bw)
        self._bandwidth = applied_bw.value
    elif tuner_set_bandwidth_supported:
        bw = int(bw)
        result = librtlsdr.rtlsdr_set_tuner_bandwidth(self.dev_p, bw)
        self._bandwidth = bw
    else:
        raise IOError('set_tuner_bandwidth not supported in this version of librtlsdr')

    if result != 0:
        self.close()
        raise LibUSBError(result, 'Could not set tuner bandwidth to %d Hz' % (requested_bw))

    return

def get_bandwidth(self):

    return getattr(self, '_bandwidth', 0)
```

--------------------------------

### Set and Get Gain with Auto/Manual Support

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Sets gain to automatic (enable AGC) or selects the nearest supported manual gain value. Retrieves the current gain in dB and uses the device gain list for nearest-match selection.

```python
def set_gain(self, gain):
    if isinstance(gain, basestring) and gain == 'auto':
        # disable manual gain -> enable AGC
        self.set_manual_gain_enabled(False)

        return

    # find supported gain nearest to one requested
    errors = [abs(10*gain - g) for g in self.gain_values]
```

--------------------------------

### Get Supported Tuner Gains in pyrtlsdr (Python)

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Fetches all supported gain values from the driver into a buffer and returns them as a list. Depends on ctypes c_int array and librtlsdr. Inputs: None; Outputs: list of int (gains in tenths of dB); Limitations: Raises IOError on failure, supports up to 50 gains.

```python
    def get_gains(self):
        """Get all supported gain values from driver

        Returns:
            list(int): Gains in tenths of a dB
        """
        buffer = (c_int *50)()
        result = librtlsdr.rtlsdr_get_tuner_gains(self.dev_p, buffer)
        if result == 0:
            self.close()
            raise IOError('Error when getting gains')

        gains = []
        for i in range(result):
            gains.append(buffer[i])

        return gains
```

--------------------------------

### Client Message Send and Get Response

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/base

Sends a client message over a socket and then waits for and returns the corresponding server response. It inherits the sending mechanism from the base class and then calls `get_response` to retrieve the server's reply.

```python
def send_message(self, sock):
        super(ClientMessage, self).send_message(sock)
        return self.get_response(sock)
```

--------------------------------

### Get Device Index by Serial in Python

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Static method to retrieve the device index for an RTL-SDR device matching a given serial number. Requires librtlsdr library for hardware interaction. Input is a serial string; output is an integer device index; raises LibUSBError if not found.

```python
[docs]    @staticmethod
    def get_device_index_by_serial(serial):
        """Retrieves the device index for a device matching the given serial number

        Arguments:
            serial (str): The serial number to search for

        Returns:
            int: The device_index as reported by ``librtlsdr``

        Notes:
            Most devices by default have the same serial number: `'0000001'`.
            This can be set to a custom value by using the `rtl\_eeprom`_ utility
            packaged with ``librtlsdr``.

        .. _rtl\_eeprom: http://manpages.ubuntu.com/manpages/trusty/man1/rtl_eeprom.1.html

        """
        if PY3 and isinstance(serial, str):
            serial = bytes(serial, 'UTF-8')

        result = librtlsdr.rtlsdr_get_index_by_serial(serial)
        if result < 0:
            raise LibUSBError(result)

        return result
```

--------------------------------

### AsyncCallbackIter Class for Asyncio Integration

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdraio

A utility class that converts legacy callback-based asynchronous functions into an asyncio-compatible iterator. It manages a queue to hold data received via callbacks and provides methods to start and stop the underlying executor task. Useful for integrating blocking I/O operations with asyncio.

```python
class AsyncCallbackIter:
    '''Convert a callback-based legacy async function into one supporting asyncio
    and Python 3.5+

    The queued data can be iterated using ``async for``

    Arguments:
        func_start: A callable which should take a single callback that will be
            passed data. Will be run in a separate thread in case it blocks.
        func_stop (optional): A callable to stop ``func_start`` from calling the
            callback. Will be run in a separate thread in case it blocks.
        queue_size (:obj:`int`, optional): The maximum amount of data
            that will be buffered.
        loop (optional): The ``asyncio.event_loop`` to use. If not supplied,
            :func:`asyncio.get_event_loop` will be used.

    '''

    def __init__(self, func_start, func_stop=None, queue_size=20, *, loop=None):
        self.queue = asyncio.Queue(queue_size)
        self.loop = loop if loop else asyncio.get_event_loop()
        self.func_stop = func_stop
        self.func_start = func_start

        self.running = False

    async def add_to_queue(self, *args):
        '''Add items to the queue

        Arguments:
            *args: Arguments to be added

        This method is a :obj:`~asyncio.coroutine`
        ''<bos>
        try:
            self.queue.put_nowait(args)
        except asyncio.QueueFull:
            log.info('extra callback data lost')

    def _callback(self, *args):
        if not self.running:
            return
        asyncio.run_coroutine_threadsafe(self.add_to_queue(*args), self.loop)

    async def start(self):
        '''Start the execution

        The callback given by ``func_start`` will be called by
        :meth:`asyncio.AbstractEventLoop.run_in_executor` and will continue
        until :meth:`stop` is called.

        This method is a :obj:`~asyncio.coroutine`
        '' '

        assert(not self.running)

        # start legacy async function
        future = self.loop.run_in_executor(None, self.func_start, self._callback)
        asyncio.ensure_future(future, loop=self.loop)
        self.executor_task = future
        self.running = True

    async def stop(self):
        '''Stop the running executor task

        If ``func_stop`` was supplied, it will be called after the queue has
        been exhausted.

        This method is a :obj:`~asyncio.coroutine`
        '' '

        assert(self.running)

        self.running = False

        # send a signal to stop
        iter_stopped = False
        while not iter_stopped:
            try:
                self.queue.put_nowait((StopAsyncIteration(),))
                iter_stopped = True
            except asyncio.QueueFull:
                try:
                    self.queue.task_done()
                except ValueError:
                    pass
        if self.func_stop:
            # stop legacy async function
            await self.loop.run_in_executor(None, self.func_stop)
        await self.executor_task

    def __aiter__(self):
        return self

    async def __anext__(self):
        val = await self.queue.get()
        self.queue.task_done()

        if isinstance(val[0], StopAsyncIteration):
            raise StopAsyncIteration

        #return val if len(val) > 1 else val[0]

        # slight hack for rtlsdr to ignore context object
        return val[0]

```

--------------------------------

### GET /api/device/frequency

Source: https://pyrtlsdr.readthedocs.io/en/latest/genindex

Gets the current center frequency of the RTLSDR device. This is essential for tuning to specific radio signals.

```APIDOC
## GET /api/device/frequency

### Description
Gets the current center frequency of the RTLSDR device.

### Method
GET

### Endpoint
/api/device/frequency

### Response
#### Success Response (200)
- **frequency** (number) - The center frequency in Hz

#### Response Example
{
  "frequency": 100000000
}
```

--------------------------------

### Get Device Serial Addresses in Python

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Static method to get serial numbers for all attached RTL-SDR devices. Depends on librtlsdr for device enumeration and USB string reading. No inputs; returns a list of strings. Limitations include reliance on device USB interfaces.

```python
[docs]    @staticmethod
    def get_device_serial_addresses():
        """Get serial numbers for all attached devices

        Returns:
            list(str): A ``list`` of all detected serial numbers (``str``)

        """
        def get_serial(device_index):
            bfr = (c_ubyte * 256)()
            r = librtlsdr.rtlsdr_get_device_usb_strings(device_index, None, None, bfr)
            if r != 0:
                raise LibUSBError(
                    r, 'while reading USB strings (device %d)' % (device_index)
                )
            return ''.join((chr(b) for b in bfr if b > 0))

        num_devices = librtlsdr.rtlsdr_get_device_count()
        return [get_serial(i) for i in range(num_devices)]
```

--------------------------------

### AsyncCallbackIter Class Initialization

Source: https://pyrtlsdr.readthedocs.io/en/latest/rtlsdraio

Shows how to initialize the AsyncCallbackIter class, which converts callback-based legacy async functions into asyncio-compatible iterators. Includes parameters for start/stop functions and queue size.

```python
class _rtlsdr.rtlsdraio.AsyncCallbackIter(_func_start_, _func_stop=None_, _queue_size=20_, _*_, _loop=None_)
```

--------------------------------

### GET /api/device/gain

Source: https://pyrtlsdr.readthedocs.io/en/latest/genindex

Retrieves the current gain setting of the RTLSDR device. This endpoint is useful for checking the amplifier gain configuration.

```APIDOC
## GET /api/device/gain

### Description
Retrieves the current gain setting of the RTLSDR device.

### Method
GET

### Endpoint
/api/device/gain

### Response
#### Success Response (200)
- **gain** (number) - The current gain value in dB

#### Response Example
{
  "gain": 20.5
}
```

--------------------------------

### Initialize RTL-SDR TCP Client with Socket Communication

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/client

Defines the RtlSdrTcpClient class constructor and core initialization. Sets up device parameters, hostname, port, and establishes initial TCP connection. The __init__ method calls open() to create the socket connection. Handles socket creation, connection establishment, and device state management.

```python
import socket

from .base import (
    CommunicationError,
    RtlSdrTcpBase,
    ClientMessage,
    ServerMessage,
    AckMessage,
    DEFAULT_READ_SIZE,
)


[docs]class RtlSdrTcpClient(RtlSdrTcpBase):

    """Client object that connects to a remote server.

    Exposes most of the methods and descriptors that are available in the
    RtlSdr class in a transparent manner allowing an interface that is nearly
    identical to the core API.

    """

    def __init__(self, device_index=0, test_mode_enabled=False,
                 hostname='127.0.0.1', port=None):
        super(RtlSdrTcpClient, self).__init__(device_index, test_mode_enabled,
                                              hostname, port)
        self.open()


[docs]    def open(self, *args):
        self._socket = None
        self._keep_alive = False
        self.device_opened = True
```

--------------------------------

### Read and print RTL-SDR samples

Source: https://pyrtlsdr.readthedocs.io/en/latest/Overview

Demonstrates basic usage of pyrtlsdr to configure an RTL-SDR device and read IQ samples. Sets sample rate, center frequency, frequency correction, and automatic gain control before reading 512 samples.

```python
from rtlsdr import RtlSdr

sdr = RtlSdr()

# configure device
sdr.sample_rate = 2.048e6  # Hz
sdr.center_freq = 70e6     # Hz
sdr.freq_correction = 60   # PPM
sdr.gain = 'auto'

print(sdr.read_samples(512))
```

--------------------------------

### RtlSdrAio Class Methods

Source: https://pyrtlsdr.readthedocs.io/en/latest/rtlsdraio

Documents the key methods of the RtlSdrAio class including stream(), stop(), and their parameters. These methods enable asynchronous sample streaming from RTL-SDR devices.

```python
stream(_num_samples_or_bytes=131072_, _format='samples'_, _loop=None_)
```

```python
stop()
```

--------------------------------

### Python: Run Server with Argument Parser

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/server

Sets up and runs the RTL-SDR TCP server with command-line arguments for hostname, port, and device index.

```python
def run_server():
    """Convenience function to run the server from the command line
    with options for hostname, port and device index.
    """
    p = argparse.ArgumentParser()
    p.add_argument(
        '-a', '--address',
        dest='hostname',
        metavar='address',
        default='127.0.0.1',
        help='Listen address (default is "127.0.0.1")')
    p.add_argument(
        '-p', '--port',
        dest='port',
        type=int,
        default=1235,
        help='Port to listen on (default is 1235)')
    p.add_argument(
        '-d', '--device-index',
        dest='device_index',
        type=int,
        default=0)
    args, remaining = p.parse_known_args()
    o = vars(args)
    server = RtlSdrTcpServer(**o)
    server.run_forever()
```

--------------------------------

### Get Server Response Class

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/base

Specifies the class that should be used to handle responses to a ServerMessage. In this implementation, it returns the AckMessage class, indicating that acknowledgments are expected.

```python
def get_response_class(self):
        return AckMessage
```

--------------------------------

### Get Server Message Header

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/base

Constructs the header for a ServerMessage. It includes success status and optionally embeds the header of the client's original request message.

```python
def get_header(self, **kwargs):
        d = super(ServerMessage, self).get_header(**kwargs)
        d['success'] = kwargs.get('success', True)
        client_message = kwargs.get('client_message')
        if client_message is not None:
            d['request'] = client_message.header
        else:
            d['request'] = kwargs.get('request')
        return d
```

--------------------------------

### BaseRtlSdr Initialization in Python

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Initializes the BaseRtlSdr instance by calling the open method. Inputs include device index, test mode flag, and serial number. No direct output; delegates to open method. Part of the class constructor.

```python
    def __init__(self, device_index=0, test_mode_enabled=False, serial_number=None):
        self.open(device_index, test_mode_enabled, serial_number)
```

--------------------------------

### Initialize RTL-SDR Device with Safe Error Handling

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Opens an RTL-SDR device by index/serial, enables test mode if requested, resets the read buffer, and initializes device values (center frequency, sample rate, gain). Validates each call to librtlsdr and raises LibUSBError on failure, closing the device on error.

```python
device_index = self.get_device_index_by_serial(serial_number)

# this is the pointer to the device structure used by all librtlsdr
# functions
self.dev_p = p_rtlsdr_dev(None)

# initialize device
result = librtlsdr.rtlsdr_open(self.dev_p, device_index)
if result < 0:
    raise LibUSBError(result, 'Could not open SDR (device index = %d)' % (device_index))

# enable test mode if necessary
result = librtlsdr.rtlsdr_set_testmode(self.dev_p, int(test_mode_enabled))
if result < 0:
    raise LibUSBError(result, 'Could not set test mode')

# reset buffers
result = librtlsdr.rtlsdr_reset_buffer(self.dev_p)
if result < 0:
    raise LibUSBError(result, 'Could not reset buffer')

self.device_opened = True
self.init_device_values()
```

--------------------------------

### Cancel ongoing async read in Python RtlSdr

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Sets the internal cancellation flag to stop any ongoing asynchronous read operation. Subsequent callbacks will be ignored until a new read is started.

```Python
def cancel_read_async(self):
    """Cancel async read."""
    self.read_async_canceling = True
```

--------------------------------

### Get ACK Response Message

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/base

Retrieves an acknowledgment (ACK) response message from a socket. This is typically used to confirm that a previously sent message has been received and processed successfully.

```python
def get_ack_response(self, sock):
        return AckMessage.from_remote(sock)
```

--------------------------------

### Open RTL-SDR Device in Python

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Method to connect to the RTL-SDR device, initialize communication, and retrieve device values. Inputs are device index, test mode flag, and serial number. No output; raises IOError on failure. Limitations include hardware dependency.

```python
[docs]    def open(self, device_index=0, test_mode_enabled=False, serial_number=None):
        """Connect to the device through the underlying wrapper library

        Initializes communication with the device and retrieves information
        from it with a call to :meth:`init_device_values`.

        Arguments:
            device_index (:obj:`int`, optional): The device index to use if there are
                multiple dongles attached.  If only one is being used,
                the default value (0) will be used.
            test_mode_enabled (:obj:`bool`, optional): If True, enables a special
                test mode, which will return the value of an internal RTL2832
                8-bit counter with calls to :meth:`read_bytes`.
            serial_number (:obj:`str`, optional): If not None, the device will be searched
                for by the given serial_number by :meth:`get_device_index_by_serial`
                and the ``device_index`` returned will be used automatically.

        Notes:
            The arguments used here are passed directly from object
            initialization.

        Raises:
            IOError: If communication with the device could not be established.

        """

        if serial_number is not None:
```

--------------------------------

### Handle Multiple RTL-SDR Devices

Source: https://pyrtlsdr.readthedocs.io/en/latest/Overview

Demonstrates how to work with multiple RTL-SDR devices by retrieving serial numbers and creating device instances based on index or serial number. Added in version 2.5.6.

```python
from rtlsdr import RtlSdr

# Get a list of detected device serial numbers (str)
serial_numbers = RtlSdr.get_device_serial_addresses()

# Find the device index for a given serial number
device_index = RtlSdr.get_device_index_by_serial('00000001')

sdr = RtlSdr(device_index)


# Or pass the serial number directly:
sdr = RtlSdr(serial_number='00000001')
```

--------------------------------

### Implement SDR Parameter Getters and Setters

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/client

Provides getter and setter methods for core SDR parameters including center frequency, sample rate, bandwidth, gain, and frequency correction. Each parameter uses the appropriate communication method (descriptor get/set or general method call) to interact with the remote server. Methods include parameter validation and error handling through the communication layer.

```python
[docs]    def get_center_freq(self):
        return self._communicate_descriptor_get('fc')



[docs]    def set_center_freq(self, value):
        self._communicate_descriptor_set('fc', value)



[docs]    def get_sample_rate(self):
        return self._communicate_descriptor_get('rs')



[docs]    def set_sample_rate(self, value):
        self._communicate_descriptor_set('rs', value)



[docs]    def get_bandwidth(self):
        return self._communicate_descriptor_get('bandwidth')



[docs]    def set_bandwidth(self, value):
        self._communicate_descriptor_set('bandwidth', value)



[docs]    def get_gain(self):
        return self._communicate_descriptor_get('gain')



[docs]    def set_gain(self, value):
        self._communicate_descriptor_set('gain', value)



[docs]    def get_freq_correction(self):
        return self._communicate_descriptor_get('freq_correction')



[docs]    def set_freq_correction(self, value):
        self._communicate_descriptor_set('freq_correction', value)
```

--------------------------------

### Get Server Message Response

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/base

Waits for and parses a specific response message from a socket. It uses the response class obtained from `get_response_class` to deserialize the message received from the remote end.

```python
def get_response(self, sock):
        """Waits for a specific response message

        The message class returned from :meth:`get_response_class` is used
        to parse the message (called from :meth:`from_remote`)

        Arguments:
            sock: The :class:`~socket.socket` object to read from
        """
        cls = self.get_response_class()
        return cls.from_remote(sock)
```

--------------------------------

### TCP Server for RTL-SDR Access

Source: https://pyrtlsdr.readthedocs.io/en/latest/Reference

This Python code sets up a TCP server using pyrtlsdr.rtlsdrtcp to allow remote access to an RTL-SDR device. Clients can connect to this server to receive I/Q samples. It requires the pyrtlsdr library and involves running a server process.

```python
from rtlsdr.rtlsdrtcp import RtlSdrTcpServer

# Configure the RTL-SDR device (e.g., sample rate, center frequency, gain)
# This will be done internally by the server when a client connects

# Create and run the TCP server
# The server will listen on '0.0.0.0' (all interfaces) and port 12345
server = RtlSdrTcpServer(num_samples=16384)
print("Starting RTL-SDR TCP server on port 12345...")
server.run_server()

```

--------------------------------

### Asynchronous Sample Reading with pyrtlsdr

Source: https://pyrtlsdr.readthedocs.io/en/latest/Reference

This Python code uses the asynchronous capabilities of pyrtlsdr to read samples from an RTL-SDR device. It sets up an asynchronous reader and a callback function to process samples as they arrive. This is useful for real-time applications where continuous data processing is needed without blocking the main thread.

```python
from rtlsdr import RtlSdr
import time

def process_samples(samples):
    # Process the received samples here
    # For example, print the first few samples:
    print(f"Received {len(samples)} samples. First 5: {samples[:5]}")

sdr = RtlSdr()
sdr.sample_rate = 2.06e6
sdr.center_freq = 100e6
sdr.gain = 'auto'

# Start asynchronous reading with a callback
sdr.read_samples_async(callback=process_samples)

print("Reading samples asynchronously. Press Ctrl+C to stop.")
try:
    while True:
        time.sleep(1) # Keep the main thread alive
except KeyboardInterrupt:
    print("Stopping...")
    sdr.cancel_read_async()
    sdr.close()

```

--------------------------------

### Set and Get Center Frequency

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Sets the device center frequency in Hz and reads it back, rounding to the nearest kHz to compensate tuner quantization (e.g., E4000). Raises LibUSBError and closes the device on failure.

```python
def set_center_freq(self, freq):

    freq = int(freq)

    result = librtlsdr.rtlsdr_set_center_freq(self.dev_p, freq)
    if result < 0:
        self.close()
        raise LibUSBError(result, 'Could not set center_freq to %d Hz' % (freq))

    return

def get_center_freq(self):

    result = librtlsdr.rtlsdr_get_center_freq(self.dev_p)
    if result < 0:
        self.close()
        raise LibUSBError(result, 'Could not get center_freq')

    # FIXME: the E4000 rounds to kHz, this may not be true for other tuners
    reported_center_freq = result
    center_freq = round(reported_center_freq, -3)

    return center_freq
```

--------------------------------

### Get Client Response Class

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/base

Defines the expected response message class for a ClientMessage. This method returns ServerMessage, indicating that clients anticipate receiving ServerMessage objects in response to their requests.

```python
def get_response_class(self):
        return ServerMessage
```

--------------------------------

### Get Client Message Header

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/base

Constructs the header for a ClientMessage. It includes common keys like 'type' and 'name', updating the header inherited from the base class with these specific client message attributes.

```python
def get_header(self, **kwargs):
        d = super(ClientMessage, self).get_header(**kwargs)
        keys = ['type', 'name']
        d.update({k: kwargs.get(k) for k in keys})
        return d
```

--------------------------------

### Connect to RTL-SDR TCP Client

Source: https://pyrtlsdr.readthedocs.io/en/latest/Overview

Demonstrates connecting to an RtlSdrTcpServer via RtlSdrTcpClient. Functions similarly to local RtlSdr instance but accesses remote hardware over network. Typically run on separate machine.

```python
# On another machine (typically)
client = RtlSdrTcpClient(hostname='192.168.1.100', port=12345)
client.center_freq = 2e6
data = client.read_samples()
```

--------------------------------

### Implement SDR Sample Reading and Advanced Configuration

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/client

Provides methods for reading samples and bytes from the SDR, including IQ sample conversion. The get_gains and get_tuner_type methods query device capabilities. Set_direct_sampling enables direct sampling mode. Includes async method placeholders that raise NotImplementedError since async operations aren't available in TCP mode.

```python
[docs]    def get_gains(self):
        return self._communicate_method('get_gains')



[docs]    def get_tuner_type(self):
        return self._communicate_method('get_tuner_type')



[docs]    def set_direct_sampling(self, value):
        self._communicate_method('set_direct_sampling', value)



[docs]    def read_bytes(self, num_bytes=DEFAULT_READ_SIZE):
        return self._communicate_method('read_bytes', num_bytes)



[docs]    def read_samples(self, num_samples=DEFAULT_READ_SIZE):
        raw_data = self._communicate_method('read_samples', num_samples)
        iq = self.packed_bytes_to_iq(raw_data)
        return iq



[docs]    def read_samples_async(self, *args):
        raise NotImplementedError('Async read not available in TCP mode')



[docs]    def read_bytes_async(self, *args):
        raise NotImplementedError('Async read not available in TCP mode')
```

--------------------------------

### Get Current Tuner Gain in pyrtlsdr (Python)

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Retrieves the current tuner gain value from the device. Depends on librtlsdr. Inputs: None; Outputs: int (gain in tenths of dB); Limitations: Raises IOError on failure, device closed on error.

```python
def get_gain(self):

        result = librtlsdr.rtlsdr_get_tuner_gain(self.dev_p)
        if 0 and result == 0:
            self.close()
            raise IOError('Error when getting gain')

        return result/10
```

--------------------------------

### Define Property Descriptors for SDR Parameters

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/client

Creates Python property descriptors for common SDR parameters to provide convenient attribute-style access. The properties map to the corresponding getter and setter methods, allowing users to access parameters like center_freq, sample_rate, gain, and freq_correction using standard Python attribute syntax. Includes both full parameter names and abbreviated aliases for compatibility.

```python

    center_freq = fc = property(get_center_freq, set_center_freq)
    sample_rate = rs = property(get_sample_rate, set_sample_rate)
    bandwidth = property(get_bandwidth, set_bandwidth)
    gain = property(get_gain, set_gain)
    freq_correction = property(get_freq_correction, set_freq_correction)
```

--------------------------------

### TCP Client for RTL-SDR Data

Source: https://pyrtlsdr.readthedocs.io/en/latest/Reference

This Python code demonstrates how to connect to an RTL-SDR TCP server using pyrtlsdr.rtlsdrtcp. The client establishes a connection, sends configuration commands (implicitly or explicitly depending on server implementation), and receives I/Q samples from the remote RTL-SDR device.

```python
from rtlsdr.rtlsdrtcp import RtlSdrTcpClient

# Connect to the TCP server
# Replace 'localhost' with the server's IP address if it's on a different machine
client = RtlSdrTcpClient(addr='localhost', port=12345)

print("Connected to RTL-SDR TCP server.")

try:
    # Read samples from the server
    # The number of samples might need to be configured or requested based on server implementation
    samples = client.read_samples()
    print(f"Received {len(samples)} samples.")
    # Process the received samples...

except Exception as e:
    print(f"An error occurred: {e}")
finally:
    client.close()
    print("Connection closed.")

```

--------------------------------

### Get Server Message Data

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/base

Retrieves the data payload for a ServerMessage. If the data is a dictionary containing a 'struct_fmt' key, it updates the message header with this format information. This is useful for handling data that needs specific packing or unpacking.

```python
def get_data(self, **kwargs):
        d = super(ServerMessage, self).get_data(**kwargs)
        if isinstance(d, dict) and 'struct_fmt' in d:
            self.header['struct_fmt'] = d['struct_fmt']
        return d
```

--------------------------------

### Message Handling Methods

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/base

Methods for creating messages from socket data and sending messages through sockets. Includes header and data processing utilities.

```Python
@classmethod
def from_remote(cls, sock):
    """Reads data from the socket and parses an instance of :class:`MessageBase`

    Arguments:
        sock: The :class:`~socket.socket` object to read from

    """
    header = cls._recv(sock)
    if not PY2:
        header = header.decode()
    kwargs = json.loads(header)
    if kwargs.get('ACK'):
        cls = AckMessage
    return cls(**kwargs)

def get_header(self, **kwargs):
    """Builds the header data for the message

    The :attr:`timestamp` is added to the header if not already present.

    Returns:
        dict:
    """
    d = {}
    ts = kwargs.get('timestamp')
    if ts is None:
        ts = time.time()
    d['timestamp'] = ts
    return d

def get_data(self, **kwargs):
    return kwargs.get('data', kwargs.get('header', {}).get('data'))

def send_message(self, sock):
    """Serializes and sends the message

    Arguments:
        sock: The :class:`~socket.socket` object to write to

    """
```

--------------------------------

### Device Property Definitions

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Property decorators defining getter/setter methods for RTL-SDR device parameters including center frequency, sample rate, gain, frequency correction, and bandwidth. Provides convenient attribute access with proper documentation.

```python
center_freq = fc = property(get_center_freq, set_center_freq,
    doc="""int: Get/Set the center frequency of the device (in Hz)""")
sample_rate = rs = property(get_sample_rate, set_sample_rate,
    doc="""int: Get/Set the sample rate of the tuner (in Hz)""")
gain = property(get_gain, set_gain,
    doc="""float or str: Get/Set gain of the tuner (in dB)

    Notes:
        If set to 'auto', AGC mode is enabled; otherwise gain is in dB.
        The actual gain used is rounded to the nearest value supported by
        the device (see the values in :attr:`valid_gains_db`).
    """)
freq_correction = property(get_freq_correction, set_freq_correction,
    doc="""int: Get/Set frequency offset of the tuner (in PPM)""")
bandwidth = property(get_bandwidth, set_bandwidth,
    doc="""int: Get/Set bandwidth value (in Hz)

    Set value to 0 (default) for automatic bandwidth selection.

    Notes:
        This value is stored locally and may not reflect the real tuner bandwidth

    """)
```

--------------------------------

### Implement RtlSdrTcpBase Class

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdrtcp/base

Base class for RTL-SDR TCP functionality. Handles device initialization and provides core methods for TCP communication with SDR devices.

```Python
class RtlSdrTcpBase(object):
    """Base class for all ``rtlsdrtcp`` functionality

    Arguments:
        device_index (:obj:`int`, optional):
        test_mode_enabled (:obj:`bool`, optional):
        hostname (:obj:`str`, optional):
        port (:obj:`int`, optional):
    """
    # Use port 1235 as default since rtl_tcp uses 1234
    DEFAULT_PORT = 1235

    def __init__(self, device_index=0, test_mode_enabled=False,
                 hostname='127.0.0.1', port=None):
        self.device_index = device_index
        self.test_mode_enabled = test_mode_enabled
        self.hostname = hostname
        self.port = port
        if self.port is None:
            self.port = self.DEFAULT_PORT
        self.device_ready = False
        self.server_thread = None
```

--------------------------------

### Stop Async Stream with pyrtlsdr

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdraio

This method stops the asynchronous stream previously started by the :meth:`stream` method. It ensures that both the sample reading and any associated executor tasks are properly terminated. It requires access to the asyncio event loop associated with the stream.

```python
def stop(self):
        """Stop async stream

        Stops the ``read_samples_async`` and ``Excecutor`` task created by
        :meth:`stream`.
        """
        return asyncio.ensure_future(self.async_iter.stop(), loop=self.async_iter.loop)
```

--------------------------------

### Set Default Device Values After Opening

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Retrieves device gain values and sets default state: center frequency, sample rate, and gain. Called after device open; populates valid gains for later selection and configures the device.

```python
def init_device_values(self):
    """Retrieves information from the device

    This method acquires the values for :attr:`gain_values`. Also sets the
    device to the default :attr:`center frequency <DEFAULT_FC>`, the
    :attr:`sample rate <DEFAULT_RS>` and :attr:`gain <DEFAULT_GAIN>`
    """
    self.gain_values = self.get_gains()
    self.valid_gains_db = [val/10 for val in self.gain_values]

    # set default state
    self.set_sample_rate(self.DEFAULT_RS)
    self.set_center_freq(self.DEFAULT_FC)
    self.set_gain(self.DEFAULT_GAIN)
```

--------------------------------

### Implement async sample reading in Python RtlSdr

Source: https://pyrtlsdr.readthedocs.io/en/latest/_modules/rtlsdr/rtlsdr

Provides asynchronous reading of complex IQ samples by wrapping read_bytes_async and converting the resulting byte stream to samples. Users supply a callback that receives the sample array and optional context.

```Python
def read_samples_async(self, callback, num_samples=DEFAULT_READ_SIZE, context=None):
    """Continuously read 'samples' from the tuner

    This is a combination of :meth:`read_samples` and :meth:`read_bytes_async`

    Arguments:
        callback: A function or method that will be called with the result.
            See :meth:`_samples_converter_callback` for the signature.
        num_samples (int): The number of samples read into each callback.
            Defaults to :attr:`DEFAULT_READ_SIZE`.
        context (Optional): Object to be passed as an argument to the callback.
            If not supplied or None, the :class:`RtlSdr` instance
            will be used.
    """
    num_bytes = 2*num_samples

    self._callback_samples = callback
    self.read_bytes_async(self._samples_converter_callback, num_bytes, context)

    return
```
Tokens:
Raw