path: root/airspyhf_waterfall.py

#!/usr/bin/python3

import os
import sys
import math

import airspyhf

import matplotlib
import numpy
import pylab
import time
import threading

import pygame
from pygame import gfxdraw

CENTER_FREQ = 101000000
SAMPLE_RATE = 912e3
SAMPLE_NUM = 2048

SCREEN_X = 1025
SCREEN_Y = 320

MOVE_STEP = int(SAMPLE_RATE/8)

sample_lock = threading.Lock()

# init AIRSPY and if no then go out
airspy = airspyhf.AirSpyHF()
if airspy.open(device_index=0) == -1:
    print("Cant open airspyhf device")
    sys.exit(1)

# config rtlsdr device
airspy.set_samplerate(int(SAMPLE_RATE))
airspy.set_frequency(CENTER_FREQ)
airspy.set_hf_agc(1)
airspy.set_hf_agc_threshold(0)
airspy.set_hf_lna(1)


def iq_abs(c):
    return (math.sqrt((c.real ** 2 + c.imag ** 2)))


# point should be normalised to 0.0 ... 1.0
def color_normalise(point):
    ret = (255, 0, 0)
    # blue
    if point < 0.0:
        point = 0.0
    if (point < 0.3):
        ret = (0, 0, int(point * 255 * 3.3))
    # yello
    elif (point < 0.7):
        ret = (0, int((point - 0.3) * 255 * 2.5), 0)
    # red
    elif (point <= 1.0):
        ret = (int((point - 0.7) * 255 * 3.3), 0, 0)
    else:
        # print "Color Error ", point
        pass
    print(ret)
    return ret


def color_mapping(x):
    "assumes -50 to 0 range, returns color"

    #r = math.fabs(int(x))
    #if r > 255:
    #    r = 255
    #print("+",r)
    r = int((x + 120) * 255 // 30)
    #r = max(0, r)
    #r = min(255, r)
    if r > 255:
        r = 255
    if r < 0:
        r = 0
    res = (r, r, r)
    #avg3 = avg(r)/3
    #if avg3 < 0.3:
    #    res = (0, 0, int(avg3 * 255 * 3.3))

    print(res)
    return res
    #r = 255-r
    #print((int(r), int(r), 100))
    #return (int(r), int(r), 100)



# def draw_Hz( surface, x, y, hz ):


arr = [[0 for i in range(0, SCREEN_X)] for j in range(0, SCREEN_Y)]

# init all pygame modules audio,video and more
pygame.init()

# [NEW] creates screen surface using constants
screen = pygame.display.set_mode((SCREEN_X, SCREEN_Y))

#samples = rtl.read_samples(SAMPLE_NUM)
sample_buffer = []
def read_samples(transfer):
    global sample_buffer

    #print("callback")
    #print("Python call back")
    t = transfer.contents
    bytes_to_write = t.sample_count * 4 * 2
    #print("Received %d samples"%(t.sample_count))
    rx_buffer = t.samples
    #print(f"{bytes_to_write} bytes receieved")
    #sample_buffer.append(rx_buffer)
    if len(sample_buffer) > SAMPLE_NUM:
        return 0
    for i in range(0,t.sample_count):
        d_re = t.samples[i].re
        d_im = t.samples[i].im
        sample_buffer.append(math.sqrt(d_re*d_re+d_im*d_im))
        #data = struct.pack("<f",d_re) # FIX ?!
        #wave_file.writeframesraw(data)
        #data = struct.pack("<f", d_im)  # FIX ?!
        #wave_file.writeframesraw(data)
    #print("End call back")
    return 0

def get_samples():
    global sample_buffer
    buf_size = len(sample_buffer)

    if buf_size < SAMPLE_NUM:
        print("sample buffer small")
        return []
    #print("getting stuff ", buf_size)
    while sample_lock.locked():
        time.sleep(0.1)
    samples = sample_buffer.copy()
    sample_buffer = [] #!
    return samples[:SAMPLE_NUM]
    #if len(samples) == num:
    #    sample_buffer = []
    #    return samples
    #if len(samples) > num:
    #    arr = []
    #    times = len(samples)/num
    #    for i in range(0, num):
    #        avg = numpy.average(samples[times*num:times*num+times-1])
    #        arr.append(avg)
    #    return arr
    #else:
    #    print("Error in get_samples")
    #    return []
    return []

read_samples_cb = airspyhf.airspyhf_sample_block_cb_fn(read_samples)
airspy.start(read_samples_cb)

run = True
line = 0
while run and airspy.is_streaming():

    print("Loop tick")

    # check for all events that where ocure
    for event in pygame.event.get():
        # if some one clicked on close button
        if event.type == pygame.QUIT:
            # terminate programm
            run = False
            # don't waste your time by waiting while event loop will end
            break
        elif event.type == pygame.KEYDOWN:
            if event.key == pygame.K_LEFT:
                print("Left")
                # rtl.center_freq -= MOVE_STEP
                airspy.set_frequency(airspy.cur_freq - MOVE_STEP)
                # print("Center freq: ", rtl.center_freq, " Hz")
            elif event.key == pygame.K_RIGHT:
                print("Right")
                # rtl.center_freq += MOVE_STEP
                # print("Center freq: ", rtl.center_freq, " Hz")
                airspy.set_frequency(airspy.cur_freq + MOVE_STEP)

    width = SCREEN_X
    height = SCREEN_Y

    samples = get_samples()
    if samples == []:
        print("sample buffer is empty")
        time.sleep(1)
        continue
    #print(samples)
    spect = numpy.fft.fft(samples,n=SAMPLE_NUM)

    spect = spect[0:int((len(spect) / 2))]

    # (1/(Fs*N)) * abs(xdft).^2;
    spect_n = [(1.0 / (SAMPLE_NUM * len(spect))) * iq_abs(x) ** 2 for x in spect]
    spect_n = (10 * numpy.log10(spect_n)).tolist()

    print(spect_n)

    # total data size
    spect_len = len(spect_n)
    #print(spect_len)
    # calculate amount spect points per pixel without rounding
    pixel_width = int(spect_len / SCREEN_X + 1)
    #print(pixel_width)
    pixel_steps = int(spect_len / pixel_width)
    #print(pixel_steps)
    #print(spect_n)
    for step in range(0, pixel_steps):
        avg = 0.0
        for i in range(0, pixel_width):
            avg += spect_n[step * pixel_width + i]
        avg /= pixel_width
        if math.isinf(avg):
            avg = -1000

        # print avg
        #gfxdraw.pixel( screen, step, line, color_normalise((100-abs(avg))/10))
        #print(int(avg))
        gfxdraw.pixel(screen, step, line, color_mapping(int(avg)))

    # draw central freq
    font = pygame.font.Font(None, 20)
    text = font.render(str(airspy.cur_freq / 1e6), 1, (200, 30, 30), (0, 0, 0))
    screen.blit(text, (SCREEN_X / 2, SCREEN_Y - 20))
    text = font.render(str((airspy.cur_freq+ SAMPLE_RATE / 2) / 1e6), 1, (200, 30, 30), (0, 0, 0))
    screen.blit(text, (SCREEN_X - 40, SCREEN_Y - 20))
    text = font.render(str((airspy.cur_freq - SAMPLE_RATE / 2) / 1e6), 1, (200, 30, 30), (0, 0, 0))
    screen.blit(text, (20, SCREEN_Y - 20))

    pygame.display.flip()
    line += 1
    if (line > SCREEN_Y):
        line = 0
    time.sleep(0.1)

pygame.quit()

airspy.stop()
airspy.close()