Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
tree: e11ad5220a
Fetching contributors…

Cannot retrieve contributors at this time

163 lines (125 sloc) 5.063 kb
from __future__ import division
import matplotlib.animation as animation
from matplotlib.mlab import psd
import pylab as pyl
import numpy as np
import sys
from rtlsdr import RtlSdr
# A simple waterfall, spectrum plotter
#
# Controls:
#
# * Scroll mouse-wheel up and down to change the center frequency (hold shift for
# finer control)
# * Press "+" and "-" to control gain, and space to enable AGC.
# * Type a frequency (in MHz) and press enter to directly change the center frequency
NFFT = 1024*4
NUM_SAMPLES_PER_SCAN = NFFT*16
NUM_BUFFERED_SWEEPS = 100
# change this to control the number of scans that are combined in a single sweep
# (e.g. 2, 3, 4, etc.) Note that it can slow things down
NUM_SCANS_PER_SWEEP = 1
# these are the increments when scrolling the mouse wheel or pressing '+' or '-'
FREQ_INC_COARSE = 1e6
FREQ_INC_FINE = 0.1e6
GAIN_INC = 5
class Waterfall(object):
keyboard_buffer = []
shift_key_down = False
image_buffer = -100*np.ones((NUM_BUFFERED_SWEEPS,\
NUM_SCANS_PER_SWEEP*NFFT))
def __init__(self, sdr=None, fig=None):
self.fig = fig if fig else pyl.figure()
self.sdr = sdr if sdr else RtlSdr()
self.init_plot()
def init_plot(self):
self.ax = self.fig.add_subplot(1,1,1)
self.image = self.ax.imshow(self.image_buffer, aspect='auto',\
interpolation='nearest', vmin=-50, vmax=10)
self.ax.set_xlabel('Current frequency (MHz)')
self.ax.get_yaxis().set_visible(False)
self.fig.canvas.mpl_connect('scroll_event', self.on_click)
self.fig.canvas.mpl_connect('key_press_event', self.on_key_press)
self.fig.canvas.mpl_connect('key_release_event', self.on_key_release)
def update_plot_labels(self):
fc = self.sdr.fc
rs = self.sdr.rs
freq_range = (fc - rs/2)/1e6, (fc + rs*(NUM_SCANS_PER_SWEEP - 0.5))/1e6
self.image.set_extent(freq_range + (0, 1))
self.fig.canvas.draw_idle()
def on_click(self, event):
if event.button == 'up':
self.sdr.fc += FREQ_INC_FINE if self.shift_key_down else FREQ_INC_COARSE
self.update_plot_labels()
elif event.button == 'down':
self.sdr.fc -= FREQ_INC_FINE if self.shift_key_down else FREQ_INC_COARSE
self.update_plot_labels()
def on_key_press(self, event):
if event.key == '+':
self.sdr.gain += GAIN_INC
elif event.key == '-':
self.sdr.gain -= GAIN_INC
elif event.key == ' ':
self.sdr.gain = 'auto'
elif event.key == 'shift':
self.shift_key_down = True
elif event.key == 'enter':
# see if valid frequency was entered, then change center frequency
try:
# join individual key presses into a string
input = ''.join(self.keyboard_buffer)
# if we're doing multiple adjacent scans, we need to figure out
# the appropriate center freq for the leftmost scan
center_freq = float(input)*1e6 + (self.sdr.rs/2)*(1 - NUM_SCANS_PER_SWEEP)
self.sdr.fc = center_freq
self.update_plot_labels()
except ValueError:
pass
self.keyboard_buffer = []
else:
self.keyboard_buffer.append(event.key)
def on_key_release(self, event):
if event.key == 'shift':
self.shift_key_down = False
def update(self, *args):
# save center freq. since we're gonna be changing it
start_fc = self.sdr.fc
# prepare space in buffer
# TODO: use indexing to avoid recreating buffer each time
self.image_buffer = np.roll(self.image_buffer, 1, axis=0)
for scan_num, start_ind in enumerate(range(0, NUM_SCANS_PER_SWEEP*NFFT, NFFT)):
self.sdr.fc += self.sdr.rs*scan_num
# estimate PSD for one scan
samples = self.sdr.read_samples(NUM_SAMPLES_PER_SCAN)
psd_scan, f = psd(samples, NFFT=NFFT)
self.image_buffer[0, start_ind: start_ind+NFFT] = 10*np.log10(psd_scan)
# plot entire sweep
self.image.set_array(self.image_buffer)
# restore original center freq.
self.sdr.fc = start_fc
return self.image,
def start(self):
self.update_plot_labels()
if sys.platform == 'darwin':
# Disable blitting. The matplotlib.animation's restore_region()
# method is only implemented for the Agg-based backends,
# which the macosx backend is not.
blit = False
else:
blit = True
ani = animation.FuncAnimation(self.fig, self.update, interval=50,
blit=blit)
pyl.show()
return
def main():
sdr = RtlSdr()
wf = Waterfall(sdr)
# some defaults
sdr.rs = 3e6
sdr.fc = 100e6
sdr.gain = 10
wf.start()
# cleanup
sdr.close()
if __name__ == '__main__':
main()
Jump to Line
Something went wrong with that request. Please try again.