gr-lora/python/qa_BasicTest.py

import lora, socket, pmt, time

from gnuradio import gr, gr_unittest, blocks
from gnuradio.blocks import socket_pdu

### Expected test results
#                    FILE                     |                 HEX
# --------------------------------------------+-------------------------------------
# hackrf_cr4-5_bw125_sf7_crc1_pwr1_001.cfile  | 40 0b 07 de ad be ef
# hackrf_cr4-5_bw125_sf7_crc1_pwr1_002.cfile  | (40 0b 07 de ad be ef) * 10
# hackrf_cr4-5_bw125_sf7_crc1_pwr1_003.cfile  | 80 0b 01 a5 a5 a5 a5 a5 a5 a5 a5
# hackrf_cr4-5_bw125_sf7_crc1_pwr1_004.cfile  | (80 0b 01 a5 a5 a5 a5 a5 a5 a5 a5) * 3

class qa_BasicTest (gr_unittest.TestCase):

    #
    #   Listen on socket for data, append to list if any and return list of captured data.
    #   
    def gatherFromSocket(self):
        timeout = 0.5
        total_data = []
        data = ''
        begin = time.time()
        while 1:
            if total_data and time.time()-begin > timeout:
                break
            elif time.time()-begin > timeout*2:
                break
            try:
                data = self.server.recv(8192)
                if data:
                    total_data.append(data)
                    begin = time.time()
                else:
                    time.sleep(0.1)
            except:
                pass

        return total_data

    #
    #   Compare captured data list with the expected data list.
    #   Print received data, expected data and if they match.
    #   Also give feedback about successrate in the decoding (percentage correctly captured and decoded).
    # 
    def compareDataSets(self, total_data, expected_data):
        total_passing = 0.0

        print '\nReceived:'

        for idx, val in enumerate(expected_data):
            data_str = '?'
            passed = True

            try:
                data_str = " ".join("{:02x}".format(ord(n)) for n in total_data[idx])
                self.assertEqual(val, data_str)
            except:
                passed = False

            if passed:
                total_passing += 1
                print '% 3d :: match:\n %s' % (idx, val)
            else:
                print '% 3d :: mismatch:\n %s' % (idx, data_str)
                print 'should be:\n %s' % (expected_data[idx])

        print '\nPassed rate: %d out of %d (%.2f%%)\n' % (total_passing, len(expected_data), total_passing / len(expected_data) * 100.0)
        
    #
    #   Connect a filesource with the given file and run the flowgraph.
    #   File is acquired from the file_sink block and saved as .cfile (complex IQ data).
    #
    def runWithFileSourceAndData(self, file, expected_data):
        print "Starting test from data in: %s\n" % (file)

        file_source = blocks.file_source(gr.sizeof_gr_complex*1, file, False) # Repeat input: True/False
        self.tb.connect((file_source, 0), (self.blocks_throttle_0, 0))
        self.tb.run ()
        self.tb.disconnect((file_source, 0), (self.blocks_throttle_0, 0))
        
        total_data = self.gatherFromSocket()
        self.compareDataSets(total_data, expected_data)

    #
    #   Set up flowgraph before Unit Test. (Gets threrefore called before EVERY test)
    # 
    #    1. Set variables for various blocks
    #    2. Make UDP socket server to listen for outputted messages
    #    3. Connect blocks: (file_source) -> throttle -> lora_receiver -> message_socket_sink
    # 
    def setUp (self):
        ##################################################
        # Variables                                      #
        ##################################################
        self.target_freq     = 868.1e6
        self.sf              = 7
        self.samp_rate       = 1e6
        self.capture_freq    = 868.0e6
        #self.bw              = 125e3
        #self.symbols_per_sec = self.bw  / (2**self.sf)
        self.offset          = -(self.capture_freq - self.target_freq)
        #self.bitrate         = self.sf * (1 / (2**self.sf / self.bw ))

        # Socket connection for sink
        self.host            = "127.0.0.1"
        self.port            = 40868

        self.server          = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.server.bind((self.host, self.port))
        self.server.setblocking(0)

        ##################################################
        # Blocks                                         #
        ##################################################
        self.tb = gr.top_block ()

        self.lora_lora_receiver_0         = lora.lora_receiver(self.samp_rate, self.capture_freq, self.offset, self.sf, self.samp_rate)
        self.blocks_throttle_0            = blocks.throttle(gr.sizeof_gr_complex*1, self.samp_rate, True)
        self.blocks_message_socket_sink_0 = lora.message_socket_sink()

        self.tb.connect(     (self.blocks_throttle_0, 0),            (self.lora_lora_receiver_0, 0))
        self.tb.msg_connect( (self.lora_lora_receiver_0, 'frames'),  (self.blocks_message_socket_sink_0, 'in'))

    #
    #   Clean-up flowgraph after Unit Test. (Gets threrefore called after EVERY test)
    # 
    def tearDown (self):
        self.tb = None
        self.server.close()

        
    ###############################################################################################
    #   Unit tests                                                                                #
    #       These assume a directory "lora-samples" with the specified .cfiles existing.          #
    ###############################################################################################
        
    #
    #   Unit test 1
    # 
    def test_001 (self):
        expected_data = [
            '71 1b 09 12 12 12 12 12 12 12 10 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12',
            '71 1b 09 88 88 88 88 88 88 88 88 88 88 88 88 88 88 88 88 88 88 88 88 88 88 88',
            '71 1b 09 00 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22'
        ]

        self.runWithFileSourceAndData('/home/william/lora-samples/usrpcr5.cfile', expected_data)

    #
    #   Unit test 2
    #
    def test_002 (self):
        expected_data = [
            '40 0b 07 de ad be ef'
        ]

        self.runWithFileSourceAndData('/home/william/lora-samples/hackrf_cr4-5_bw125_sf7_crc1_pwr1_001.cfile', expected_data)

    #
    #   Unit test 3
    #
    def test_003 (self):
        expected_data = [
            '40 0b 07 de ad be ef'
        ] * 10

        self.runWithFileSourceAndData('/home/william/lora-samples/hackrf_cr4-5_bw125_sf7_crc1_pwr1_002.cfile', expected_data)

    #
    #   Unit test 4
    #
    def test_004 (self):
        expected_data = [
            '80 0b 01 a5 a5 a5 a5 a5 a5 a5 a5'
        ]

        self.runWithFileSourceAndData('/home/william/lora-samples/hackrf_cr4-5_bw125_sf7_crc1_pwr1_003.cfile', expected_data)

    #
    #   Unit test 5
    #
    def test_005 (self):
        expected_data = [
            '80 0b 01 a5 a5 a5 a5 a5 a5 a5 a5'
        ] * 3

        self.runWithFileSourceAndData('/home/william/lora-samples/hackrf_cr4-5_bw125_sf7_crc1_pwr1_004.cfile', expected_data)

if __name__ == '__main__':
    # Remove second argument, otherwise this would cause each test to run twice 
    # (once to console and once to xml, both outputs are printed in the console though)
    gr_unittest.run(qa_BasicTest)  # , "qa_BasicTest.xml")