<jupyter_start><jupyter_text>AFSK Demodulator Step 4: Low Pass 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_04" 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_04.bit')
dma = overlay.demodulator.dma<jupyter_output><empty_output><jupyter_text>Accellerating FIR FiltersBelow is the implementation of the AFSK demodulator in Python.  We are now going to remove the low 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

block_size = 2640

xlnk = Xlnk()

def demod(data):
    start_time = time.time()
    output = np.array([],dtype=np.bool)
    with xlnk.cma_array(shape=(block_size,), dtype=np.int16) as out_buffer, \
        xlnk.cma_array(shape=(block_size,), dtype=np.int8) as in_buffer:

        for i in range(0, len(data), block_size):
            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('FPGA demodulator execution time: ',sw_exec_time)
    return output

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]
delay = 12 # ~446us

bpf_delay = 70
lpf_delay = 50
filter_delay = bpf_delay + lpf_delay

# demodulate the audio data
d = demod(audio_data[:26400])

# like before, the sign has changed.  We need to revert that before it goes into the PLL
dx = np.append(d, demod(np.zeros(filter_delay)))[filter_delay:] * -1

print(dx[:16], 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
for i in range(len(dx)):
    sample[i] = pll(dx[i])
    locked[i] = pll.locked()
    
nrzi = NRZI()

data = [int(nrzi(x)) for x,y in zip(dx, sample) if y]

hdlc = HDLC()

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

# xlnk.xlnk_reset()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1098186794389326, 'fraction_numerical': 0.01847417037290455, 'mean_word_length': 3.5123456790123457, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, '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': '146223', 'license': 'MIT', 'n_tokens_mistral': 1079, 'n_tokens_neox': 1024, 'n_words': 291, 'path': 'low_pass_filter/Pynq/afsk-demodulator-pynq.ipynb', 'repo_name': 'mobilinkd/afsk-demodulator-pynq', 'size': '2923'}