Copy

applications/unb1_fn_bf/tb/python/tc_unb1_fn_bf.py

+#! /usr/bin/env python
+###############################################################################
+#
+# Copyright (C) 2012
+# ASTRON (Netherlands Institute for Radio Astronomy) <http://www.astron.nl/>
+# P.O.Box 2, 7990 AA Dwingeloo, The Netherlands
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+###############################################################################
+
+"""Test case for the fn_bf design
+
+   Usage: python tc_fn_bf.py --unb 0 --fn 0:3 --rep 1 -n 1 -v 6
+
+"""
+
+###############################################################################
+# System imports
+import sys
+import test_case
+import node_io
+import pi_unb_sens
+import pi_diag_block_gen
+import pi_bf_bf
+import pi_st_sst
+import pi_ss_ss_wide
+import unb_apertif as apr
+from tools import *
+from common import *
+
+# Create a test case object
+tc = test_case.Testcase('TB - ', '')
+tc.set_result('PASSED')
+tc.append_log(3, '>>>')
+tc.append_log(1, '>>> Title : Test case for the fn_bf design on %s' % tc.unb_nodes_string())
+tc.append_log(3, '>>>')
+tc.append_log(3, '')
+
+# Constants/Generics that are shared between VHDL and Python
+# Name                   Value   Default   Description
+# START_VHDL_GENERICS
+c_nof_signal_paths     = 16      # 64
+c_nof_input_streams    =  4      # 16 
+c_nof_subbands         = 24      # 24
+c_nof_weights          = 256      # 256
+c_nof_bf_units         =  1      # 4
+c_in_dat_w             = 16      # 16
+c_in_weight_w          = 16      # 16
+c_gain_w               = -1      # -1
+c_bst_dat_w            = 16      # 16                                
+c_bst_lsb_w            = 14      # 14
+c_out_dat_w            = 39      # 8 for the correlator, 16 for udp_offload, 39 for capturein databuffer of testbench
+c_out_lsb_w            = 14      # 14
+c_stat_data_w          = 56      # 56
+c_stat_data_sz         =  2      # 2
+c_bf_weights_file_name = "../../../build/data/weights" #  -- "UNUSED" or relative path to e.g. the bf/build/data/weights hex file for adr_w=8 and dat_w=32
+# END_VHDL_GENERICS    
+
+c_blocks_per_sync      = 100
+c_clk_period           = 1  # ns
+# Overrule constant definitions when testcase runs on hardware. 
+if tc.sim==False:
+    c_nof_signal_paths     = 64
+    c_nof_input_streams    = 16 
+    c_nof_subbands         = 24
+    c_nof_weights          = 256
+    c_nof_bf_units         = 4
+    c_in_dat_w             = 16
+    c_in_weight_w          = 16
+    c_gain_w               = -1
+    c_bst_dat_w            = 16                                
+    c_bst_lsb_w            = 14
+    c_out_dat_w            = 39  # 8 for the correlator, 16 for udp_offload, 39 for capturein databuffer of testbench
+    c_out_lsb_w            = 14
+    c_stat_data_w          = 56
+    c_stat_data_sz         = 2
+    c_bf_weights_file_name = "../../../build/data/weights" #  -- "UNUSED" or relative path to e.g. the bf/build/data/weights hex file for adr_w=8 and dat_w=32
+    c_blocks_per_sync      = 781250
+
+c_nof_sp_per_input_stream = c_nof_signal_paths / c_nof_input_streams
+
+# Define settings for the block generator
+c_samples_per_packet = c_nof_sp_per_input_stream * c_nof_subbands
+c_gapsize            = c_nof_weights - c_samples_per_packet
+c_mem_low_addr       = 0
+c_mem_high_addr      = c_samples_per_packet-1
+c_bsn_init           = 42
+    
+# Create access object for all nodes
+io = node_io.NodeIO(tc.nodeImages, tc.base_ip)
+
+# Create instances for the block generator (BG)
+bg = pi_diag_block_gen.PiDiagBlockGen(tc, io, nofChannels=c_nof_input_streams, ramSizePerChannel= 4*2**(ceil_log2(c_samples_per_packet)))
+
+# Create instances for the beamformer units (BF)
+bf=[]   
+for i in range(tc.nofFnNodes):
+    for j in xrange(c_nof_bf_units):
+        bf.append(pi_bf_bf.PiBfBf(tc, io, c_nof_weights, c_nof_signal_paths, c_nof_input_streams, xstEnable=True, instanceNr=j, nodeNr=tc.nodeFnNrs[i]))
+                                                          
+# Set the treshold register of the statisticsmodules to zero.                                                           
+for k in xrange(tc.nofFnNodes):   
+    for i in range(c_nof_bf_units):
+        print bf[k*c_nof_bf_units+i].st.write_treshold([0])
+n=0
+                                                           
+for rep in xrange(tc.repeat):
+    
+    tc.append_log(3, '>>> Rep %d' % rep)
+
+    ################################################################################
+    ##
+    ## Initialize the blockgenerators
+    ##
+    ################################################################################
+    # - Write settings to the block generator 
+    tc.append_log(3, '>>>')
+    tc.append_log(3, '>>> Write settings to the block generator')
+    tc.append_log(3, '>>>')
+    bg.write_block_gen_settings(c_samples_per_packet, c_blocks_per_sync, c_gapsize, c_mem_low_addr, c_mem_high_addr, c_bsn_init)
+    
+    # - Create a list with the input data and write it to the RAMs of the block generator
+    tc.append_log(3, '>>>')
+    tc.append_log(3, '>>> Write data to the waveform RAM of all channels')
+    tc.append_log(3, '>>>')
+
+    inputData = []
+    for i in xrange(c_nof_input_streams):
+        dataList = bg.generate_data_list(c_nof_sp_per_input_stream, c_nof_subbands, 2048*i*4, i, c_in_dat_w)
+        print dataList
+        print ""
+        bg.write_waveform_ram(dataList, i)
+        dataListComplex = bg.convert_concatenated_to_complex(dataList, c_in_dat_w)
+        inputData.append(dataListComplex)
+        print dataListComplex
+    ################################################################################
+    ##
+    ## Create and Write the weight factors 
+    ##
+    ################################################################################
+    tc.append_log(3, '>>>')
+    tc.append_log(3, '>>> Create and write weightfactors for all signal paths on all bf_units ')
+    tc.append_log(3, '>>>')
+   
+    weightsNodes = []
+    for k in xrange(tc.nofFnNodes):                    
+        weightsBf = []
+        for i in range(c_nof_bf_units):
+            weightsBfUnit=[]
+            for j in range(c_nof_signal_paths):
+                weightsSignalPath = bf[k*c_nof_bf_units+i].generate_weights(i+j, i, c_in_weight_w)
+                bf[k*c_nof_bf_units+i].write_weights(weightsSignalPath, j) 
+                weightsSignalPathComplex = bg.convert_concatenated_to_complex(weightsSignalPath, c_in_weight_w)
+                weightsBfUnit.append(weightsSignalPathComplex)
+            weightsBf.append(weightsBfUnit)
+        weightsNodes.append(weightsBf)
+    
+    ################################################################################
+    ##
+    ## Create and Write the selection buffers
+    ##
+    ################################################################################
+    select_buf = []
+    for i in range(c_nof_sp_per_input_stream): 
+        select_buf_line = []
+        for j in range(4):
+            for k in range(c_nof_weights/4): 
+                select_buf_line.append(i*c_nof_subbands + j)  
+        select_buf.append(select_buf_line)
+
+    for i in range(tc.nofFnNodes):
+        for j in xrange(c_nof_bf_units):
+            for k in range(c_nof_input_streams):
+                bf[i*c_nof_bf_units + j].ss_wide[k].write_selects(flatten(select_buf)); 
+
+    # - Enable the block generator
+    tc.append_log(3, '>>>')
+    tc.append_log(3, '>>> Enable the block generator')
+    tc.append_log(3, '>>>')
+    tc.append_log(3, '')
+    bg.write_enable()
+    
+    ################################################################################
+    ##
+    ## Calculate the reference values
+    ##
+    ################################################################################
+    statisticsAccumulatedNode = []
+    for k in xrange(tc.nofFnNodes):   
+        statisticsAccumulated = []                 
+        for i in range(c_nof_bf_units):
+            statisticsAccumulated.append(bf[k*c_nof_bf_units+i].calculate_beamlets(inputData, select_buf, weightsNodes[k][i], c_in_weight_w, c_blocks_per_sync))
+        statisticsAccumulatedNode.append(statisticsAccumulated)
+
+    ################################################################################
+    ##
+    ## Read out the beamlet statistics 
+    ##
+    ################################################################################
+    # Wait a while before reading out the statistics
+    if tc.sim == True: 
+        current_time = io.simIO.getSimTime()
+        wait_time = current_time[0] + 2*c_blocks_per_sync * c_nof_weights * c_clk_period
+        do_until_gt(io.simIO.getSimTime, wait_time, s_timeout=3600)
+    else:
+        tc.sleep(1)
+        
+    tc.append_log(3, '>>>')
+    tc.append_log(2, '>>> Rep = %d, n = %d: Read the Beamlet Statistics of all bf_units' % (rep, n))
+    tc.append_log(3, '>>>')
+    tc.append_log(3, '')
+    
+    for k in xrange(tc.nofFnNodes):   
+        beamlet_stats = []
+        for i in range(c_nof_bf_units):
+            beamlet_stats_bf_unit = bf[k*c_nof_bf_units+i].st.read_and_verify_stats(statisticsAccumulatedNode[k][i])
+            beamlet_stats.append(beamlet_stats_bf_unit)
+        
+    # - Disable the block generator
+    tc.append_log(3, '>>>')
+    tc.append_log(3, '>>> Disable the block generator')
+    tc.append_log(3, '>>>')
+    tc.append_log(3, '')
+    bg.write_disable()
+        
+    n+=1
+
+    
+###############################################################################
+# End
+tc.set_section_id('')
+tc.append_log(3, '')
+tc.append_log(3, '>>>')
+tc.append_log(0, '>>> Test bench result: %s' % tc.get_result())
+tc.append_log(3, '>>>')
+
+sys.exit(tc.get_result())
\ No newline at end of file