<jupyter_start><jupyter_text>AFSK Demodulator Step 2: Band-Pass FIR FilterThis is a Pynq portion of the AFSK demodulator project.  We will be using the FPGA overlay that we created in Vivado.At this point we have created the bitstream for "project_02" and copied the bitstream, TCL wrapper, and hardware hand-off file to the Pynq board.Let's first verify that we can load the module.<jupyter_code>from pynq import Overlay, Xlnk
import numpy as np
import pynq.lib.dma

overlay = Overlay('project_02.bit')
dma = overlay.bpfilter.bpf_dma<jupyter_output><empty_output><jupyter_text>Accellerating FIR FilterBelow is the implementation of the AFSK demodulator in Python.  We are going to remove the band pass filter code and replace it with new code.<jupyter_code>import sys
sys.path.append('../../base')

import numpy as np
from scipy.signal import lfiltic, lfilter, firwin
from scipy.io.wavfile import read
from DigitalPLL import DigitalPLL
from HDLC import HDLC
from AX25 import AX25
import time

from pynq import Overlay, Xlnk
import numpy as np

block_size = 1024*1024

xlnk = Xlnk()
out_buffer = xlnk.cma_array(shape=(block_size,), dtype=np.int16)
in_buffer = xlnk.cma_array(shape=(block_size,), dtype=np.int16)

def bpf(data):
    start_time = time.time()
    output = np.array([],dtype=np.int16)
    for i in range(0, len(data), block_size):
        print(i)
        out_buffer[:len(data[i:i+block_size])] = data[i:i+block_size]
        dma.sendchannel.transfer(out_buffer)
        dma.recvchannel.transfer(in_buffer)
        dma.sendchannel.wait()
        dma.recvchannel.wait()
        output = np.append(output, in_buffer)
    stop_time = time.time()
    sw_exec_time = stop_time - start_time
    print('Hardware FIR execution time: ',sw_exec_time)
    return output

class fir_filter(object):
    def __init__(self, coeffs):
        self.coeffs = coeffs
        self.zl = lfiltic(self.coeffs, 32768.0, [], [])
    def __call__(self, data):
        result, self.zl = lfilter(self.coeffs, 32768.0, data, -1, self.zl)
        return result

class NRZI:

    def __init__(self):

        self.state = False

    def __call__(self, x):
        
        result = (x == self.state)
        self.state = x
        return result


audio_file = read('../../base/TNC_Test_Ver-1.102-26400-1sec.wav')
sample_rate = audio_file[0]
audio_data = audio_file[1]
print(len(audio_data))
delay = 12 # ~446us

lpf_coeffs = np.array(firwin(101, [1200.0/(sample_rate/2)], width = None,
        pass_zero = True, scale = True, window='hann') * 32768, dtype=int)

lpf = fir_filter(lpf_coeffs)

filter_delay = 64 + 50

# Band-pass filter the audio data
f = bpf(np.append(audio_data, np.zeros(filter_delay, dtype=int)))
# Digitize the data
print("Digitizing audio data...")
print(len(f))
d = np.greater_equal(f, 0, dtype=int)
print(len(d))
# Delay the data
print("Delay...")
a = d[delay:]
# XOR the digitized data with the delayed version
print("Doing Logical XOR...")
x = np.logical_xor(d[:0-delay], a, dtype=int)
# Low-pass filter the PWM signal
print("Doing LPF...")
c = lpf(x-0.5)
# Digitize the tone transistions
print(len(c))
print("Digitizing correlator output...")
dx = np.greater_equal(c, 0.0)
print(len(dx))
# Create the PLL
pll = DigitalPLL(sample_rate, 1200.0)

locked = np.zeros(len(dx), dtype=int)
sample = np.zeros(len(dx), dtype=int)

# Clock recovery
print("Doing clock recovery...")
for i in range(len(dx)):
    sample[i] = pll(dx[i])
    locked[i] = pll.locked()
    
nrzi = NRZI()

print("Doing NRZI...")
data = [int(nrzi(x)) for x,y in zip(dx, sample) if y]

hdlc = HDLC()

print("Doing HDLC")
count = 0
for b,s,l in zip(dx, sample, locked):
    if s:
        packet = hdlc(nrzi(b), l)
        if packet is not None:
            count += 1
            print(count, AX25(packet[1]))

ctrl = dma.recvchannel._mmio.read(dma.recvchannel._offset)
print(ctrl)
dma.recvchannel._mmio.write(dma.recvchannel._offset, (ctrl | 4) & 0xFFFFFFFE)
print(dma.recvchannel._mmio.read(dma.recvchannel._offset+0x04))
dma.recvchannel.start()
dma.sendchannel.start()
xlnk.xlnk_reset()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12912692589875274, 'fraction_numerical': 0.022743947175348497, 'mean_word_length': 3.880668257756563, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '457562', 'license': 'MIT', 'n_tokens_mistral': 1568, 'n_tokens_neox': 1487, 'n_words': 386, 'path': 'band_pass_filter/Pynq/afsk-demodulator-pynq.ipynb', 'repo_name': 'mobilinkd/afsk-demodulator-pynq', 'size': '4089'}