Skip to content
Snippets Groups Projects
Commit 39437f83 authored by Daniel van der Schuur's avatar Daniel van der Schuur
Browse files

-Moved functions from verify_correlator to new file_io.py and common_dsp.py.

parent da193f4b
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
libraries/dsp/correlator/tb/python/verify_correlator.py
+ 86
− 291
View file @ 39437f83
...@@ -23,6 +23,7 @@ import os ...@@ -23,6 +23,7 @@ import os
import time import time
from common import * from common import *
from common_dsp import * from common_dsp import *
from file_io import *
from mem_init_file import list_to_hex from mem_init_file import list_to_hex
# Purpose: # Purpose:
...@@ -38,21 +39,26 @@ from mem_init_file import list_to_hex ...@@ -38,21 +39,26 @@ from mem_init_file import list_to_hex
# . .. # . ..
# . Input 23 - 23 degrees # . Input 23 - 23 degrees
NOF_INPUTS = 6 NOF_INPUTS = 7
NOF_CHANNELS = 4 NOF_CHANNELS = 8
COMPLEX_WIDTH = 8 COMPLEX_WIDTH = 8
NOF_FOLDS = 0 NOF_PRE_MULT_FOLDS = 2
NOF_INPUT_FOLDS = 0 # =< NOF_PRE_MULT_FOLDS
USE_NEW_REC_FILE = False # True = don't use outdated .rec file; wait for new one.
PLOT_CORR_OUT_PHASES = False # Plot the correlator output phases of channel 0 of both the VHDL and the Model USE_NEW_REC_FILE = False #True = don't use outdated .rec file; wait for new one.
PLOT_CORR_OUT_PHASES = True # Plot the correlator output phases of channel 0 of both the VHDL and the Model
NOF_VISIBILITIES = (NOF_INPUTS*(NOF_INPUTS+1))/2 NOF_VISIBILITIES = (NOF_INPUTS*(NOF_INPUTS+1))/2
NOF_LINES_IN_REC_FILE_TO_VERIFY = NOF_CHANNELS*NOF_VISIBILITIES NOF_LINES_IN_REC_FILE_TO_VERIFY = NOF_CHANNELS*NOF_VISIBILITIES
NOF_INPUT_STREAMS = NOF_INPUTS/pow(2, NOF_FOLDS) NOF_INPUT_STREAMS = NOF_INPUTS/pow(2, NOF_INPUT_FOLDS)
INTEGRATION_PERIOD = NOF_VISIBILITIES
NOF_MULTS = ceil_div(NOF_VISIBILITIES, pow(2, NOF_PRE_MULT_FOLDS))
INTEGRATION_PERIOD = NOF_MULTS # We always use the minimum integration period needed to be able to interleave the accumulator outputs onto one stream
CORRELATOR_MULT_OUTPUT_COMPLEX_WIDTH = 2* COMPLEX_WIDTH CORRELATOR_MULT_OUTPUT_COMPLEX_WIDTH = 2* COMPLEX_WIDTH
CORRELATOR_OUTPUT_COMPLEX_WIDTH = CORRELATOR_MULT_OUTPUT_COMPLEX_WIDTH + ceil_log2(INTEGRATION_PERIOD) CORRELATOR_OUTPUT_COMPLEX_WIDTH = CORRELATOR_MULT_OUTPUT_COMPLEX_WIDTH + ceil_log2(INTEGRATION_PERIOD)
NOF_WORDS_PER_BLOCK = NOF_CHANNELS*pow(2, NOF_FOLDS) NOF_WORDS_PER_BLOCK = NOF_CHANNELS*pow(2, NOF_INPUT_FOLDS)
MEM_WIDTH = COMPLEX_WIDTH*2 MEM_WIDTH = COMPLEX_WIDTH*2
MEM_DEPTH = NOF_WORDS_PER_BLOCK MEM_DEPTH = NOF_WORDS_PER_BLOCK
PATH = "../../src/hex" PATH = "../../src/hex"
...@@ -60,262 +66,29 @@ FILENAME = "complex_subbands" ...@@ -60,262 +66,29 @@ FILENAME = "complex_subbands"
REC_FILE = os.environ['RADIOHDL']+'/libraries/dsp/correlator/tb/rec/correlator_src_out_arr0.rec' REC_FILE = os.environ['RADIOHDL']+'/libraries/dsp/correlator/tb/rec/correlator_src_out_arr0.rec'
NOF_DP_RECORD_FIELDS = 11
DP_RECORD_INDEX_SYNC = 0
DP_RECORD_INDEX_BSN = 1
DP_RECORD_INDEX_DATA = 2
DP_RECORD_INDEX_RE = 3
DP_RECORD_INDEX_IM = 4
DP_RECORD_INDEX_VALID = 5
DP_RECORD_INDEX_SOP = 6
DP_RECORD_INDEX_EOP = 7
DP_RECORD_INDEX_EMPTY = 8
DP_RECORD_INDEX_CHANNEL = 9
DP_RECORD_INDEX_ERR = 10
def gen_complex_inputs(nof_inputs, nof_channels, complex_width):
"""
Returns a 2d list of NOF_INPUTS*NOF_CHANNELS, in complex format.
. 1 deg. phase shift between the inputs
. Amplitude increases with increasing channel index
"""
phase_incr = 1
ampl_incr = 1
# != amplitude maar Y-val. Mag dus ook negatief zijn.
max_amplitude = pow(2, complex_width)/2-1
ampl_lo = max_amplitude-nof_channels*ampl_incr
input_lists = []
phase_deg = 0
for input_nr in range(nof_inputs):
channel_block = [] # one channel block carries NOF_CHANNELS channel samples
ampl = ampl_lo
phase_deg = phase_deg + phase_incr
for channel_nr in range(nof_channels):
ampl = ampl + ampl_incr
print 'input_nr',input_nr, 'channel_nr', channel_nr, 'ampl',ampl,'phase_deg', phase_deg
sample = complex_phasor_to_binomial(ampl, phase_deg)
channel_block.append(sample)
input_lists.append(channel_block)
return input_lists
def complex_to_int(input_lists):
"""
complex to int
"""
output_lists = []
for input_nr in range(NOF_INPUTS):
output_list = []
for word in range(NOF_CHANNELS):
re = int(round(input_lists[input_nr][word].real))
im = int(round(input_lists[input_nr][word].imag))
print 'Input', input_nr, 'Channel', word, 're,im', re, im
output_list.append(complex(re,im))
output_lists.append(output_list)
return output_lists #correlator_snk_in_arr
def concat_complex(input_lists):
"""
Concatenate im&re into one integer
"""
output_lists = []
for input_nr in range(NOF_INPUTS):
output_list = []
for word in range(NOF_CHANNELS):
re_bits = CommonBits(int(input_lists[input_nr][word].real), COMPLEX_WIDTH)
im_bits = CommonBits(int(input_lists[input_nr][word].imag), COMPLEX_WIDTH)
concat_bits = im_bits & re_bits
output_list.append(concat_bits.data)
output_lists.append(output_list)
return output_lists
def gen_correlator_snk_in_arr_hex(input_lists): def gen_correlator_snk_in_arr_hex(input_lists):
""" """
bla bla
""" """
concat_input_lists = concat_complex(input_lists) concat_input_lists = concat_complex_2arr(input_lists, NOF_INPUTS, NOF_CHANNELS, COMPLEX_WIDTH)
# ========================================================== # ==========================================================
# Interleave the lists if user wants folded correlator input # Interleave the lists if user wants folded correlator input
# . Note: no support for more than 1 fold yet. # . Note: no support for more than 1 fold yet.
# ========================================================== # ==========================================================
if NOF_FOLDS==1: if NOF_INPUT_FOLDS==1:
input_stream_lists = [] input_stream_lists = []
for input_stream_nr in range(NOF_INPUT_STREAMS): for input_stream_nr in range(NOF_INPUT_STREAMS):
input_stream_lists.append( interleave([concat_input_lists[2*input_stream_nr], concat_input_lists[2*input_stream_nr+1]] ) ) input_stream_lists.append( interleave([concat_input_lists[2*input_stream_nr], concat_input_lists[2*input_stream_nr+1]] ) )
elif NOF_FOLDS==0: elif NOF_INPUT_FOLDS==0:
input_stream_lists = concat_input_lists input_stream_lists = concat_input_lists
# ==================== # ====================
# Write the HEX files # Write the HEX files
# ==================== # ====================
for input_stream_nr in range(NOF_INPUT_STREAMS): for input_stream_nr in range(NOF_INPUT_STREAMS):
list_to_hex( input_stream_lists[input_stream_nr], PATH+"/"+FILENAME+"_"+str(COMPLEX_WIDTH)+'b_fold_'+str(NOF_FOLDS)+'_'+str(input_stream_nr)+".hex", MEM_WIDTH, MEM_DEPTH) list_to_hex( input_stream_lists[input_stream_nr], PATH+"/"+FILENAME+"_"+str(COMPLEX_WIDTH)+'b_fold_'+str(NOF_INPUT_FOLDS)+'_'+str(input_stream_nr)+".hex", MEM_WIDTH, MEM_DEPTH)
def complex_float_to_int(complex_list_float):
"""
Convert a list of float complex to a list of integer complex.
"""
complex_list_int = []
for word in complex_list_float:
re_int = int(round(word.real))
im_int = int(round(word.imag))
complex_list_int.append( complex(re_int, im_int) )
return complex_list_int
def correlate(input_lists, accumulation_factor):
"""
Input_lists: nof_inputs*nof_channels
Output: nof_channels*nof_visibilities
"""
# First transpose the lists into nof_channels*nof_inputs
inputs_per_channel = transpose(input_lists)
# Feed each channel to the correlation function
vis_per_channel = []
for inputs in inputs_per_channel:
vis = complex_matrix_corr(inputs, inputs, 'complex', conjugate=True)
# complex_matrix_corr() returns a full matrix of nof_inputs*nof_inputs.
# Convert it to only the triangle incl.auto correlations - so don't
# include duplicates.
unique_vis_per_channel = full_matrix_to_unique_vis(vis)
# Convert the compelx floats to complex integers
unique_vis_per_channel_int = complex_float_to_int(unique_vis_per_channel)
# Fake accumulation by multiplying by accumulation_factor
# . this is OK because the inputs do not change in time.
unique_vis_per_channel_int_acc = []
for v in unique_vis_per_channel_int:
unique_vis_per_channel_int_acc.append(accumulation_factor*v)
vis_per_channel.append( unique_vis_per_channel_int_acc )
return vis_per_channel
def rec_file_to_list(filename, field_indices=range(11)):
"""
Read a .rec file written by dp_stream_recorder.vhd and return the contents
as strings in a Python list.
Every line in the file has 11 space-seperated fields:
0 1 2 3 4 5 6 7 8 9 10
[sync] [bsn] [data] [re] [im] [valid] [sop] [eop] [empty] [channel] [err]
A list of length 0..nof_lines-1 is returned for each field passed in
field_indices.
"""
fields_sync = []
fields_bsn = []
fields_data = []
fields_re = []
fields_im = []
fields_valid = []
fields_sop = []
fields_eop = []
fields_empty = []
fields_channel = []
fields_err = []
################################################################################
# Read the lines from the file and turn them into a list of lines
################################################################################
with open (filename, "r") as recfile:
lines = recfile.read().splitlines()
# Split each line into the 11 DP fields
for line in lines:
split_line = line.split(' ')
# Only process complete lines
if len(split_line)==NOF_DP_RECORD_FIELDS:
fields_sync.append ( split_line[DP_RECORD_INDEX_SYNC] )
fields_bsn.append ( split_line[DP_RECORD_INDEX_BSN] )
fields_data.append ( split_line[DP_RECORD_INDEX_DATA] )
fields_re.append ( split_line[DP_RECORD_INDEX_RE] )
fields_im.append ( split_line[DP_RECORD_INDEX_IM] )
fields_valid.append ( split_line[DP_RECORD_INDEX_VALID] )
fields_sop.append ( split_line[DP_RECORD_INDEX_SOP] )
fields_eop.append ( split_line[DP_RECORD_INDEX_EOP] )
fields_empty.append ( split_line[DP_RECORD_INDEX_EMPTY] )
fields_channel.append( split_line[DP_RECORD_INDEX_CHANNEL])
fields_err.append ( split_line[DP_RECORD_INDEX_ERR] )
fields = [fields_sync,fields_bsn,fields_data,fields_re,fields_im,fields_valid,fields_sop,fields_eop,fields_empty,fields_channel,fields_err]
return [fields[i] for i in field_indices]
def wait_for_rec_file(filename, nof_lines):
"""
Wait for .rec file with filename to contain nof_lines.
"""
cur_time = datetime.datetime.now()
print cur_time
def file_a_newer_than_b(filename_a, filename_b):
"""
Returns True if file_a is newer than file_b, else False.
"""
file_a_mtime = os.path.getmtime(filename_a)
file_a_mtime = os.path.getmtime(filename_b)
if file_a_mtime > file_b_mtime:
return True
else:
return False
def file_mod_time(filename):
"""
Returns the time a file was last modified.
"""
return os.path.getmtime(filename)
def file_linecount(filename):
"""
Returns the number of lines in file with fname.
"""
with open(filename) as f:
return sum(1 for _ in f)
def rec_file_to_complex(filename, complex_width):
"""
Read from a .rec file and extract the complex fields.
Returns complex format.
The complex_list is re-shaped into packet-sized (as tagged by SOP and EOP)
sublists.
"""
# Read REC file and extract re,im,EOP fields.
re_im_eop_strings = rec_file_to_list(filename, [DP_RECORD_INDEX_RE,DP_RECORD_INDEX_IM,DP_RECORD_INDEX_EOP])
complex_list = []
# Convert re,im strings to complex format
for re_str,im_str in zip(re_im_eop_strings[0], re_im_eop_strings[1]):
re = to_signed(int(re_str, 16), complex_width)
im = to_signed(int(im_str, 16), complex_width)
# Convert integers to complex
complex_list.append( complex(re,im) )
# Figure out the packet size based on EOP
for word_nr,eop in enumerate(re_im_eop_strings[2]):
if eop == '1':
packet_size=word_nr+1
break
# Split the complex_list into packet_size chunks and return the result.
return split_list(complex_list, packet_size)
def compare_correlator_src_out_arr(vhdl_out, model_out): def compare_correlator_src_out_arr(vhdl_out, model_out):
""" """
...@@ -335,65 +108,87 @@ def compare_correlator_src_out_arr(vhdl_out, model_out): ...@@ -335,65 +108,87 @@ def compare_correlator_src_out_arr(vhdl_out, model_out):
channel_nr+=1 channel_nr+=1
print 'Checked %d channels, %d visibilities each: PASSED' %(channel_nr, vis_nr) print 'Checked %d channels, %d visibilities each: PASSED' %(channel_nr, vis_nr)
def plot_phase_shifts(correlator_output):
def run_tb():
""" """
""" """
################################################################################ import subprocess
# Extract the phases and amplitudes import node_io
################################################################################ import test_case
phases = []
amplitudes = []
for word in correlator_output:
phasor = complex_binomial_to_phasor(word)
ampl = phasor[0]
amplitudes.append(ampl)
phase = phasor[1]
phases.append(phase)
################################################################################
# Split the list into chunks of nof_visibilities, plot them channel after channel
################################################################################
phases_per_channel = split_list(phases, NOF_VISIBILITIES)
for channel_phases in phases_per_channel:
phases_mat = numpy.array(channel_phases)
mat = unique_vis_to_full_matrix(phases_mat)
plot_matrix_color([[mat]])
# Create a .do file for ModelSim
c_do_file_name = os.environ['HOME']+"/.ms_do.do"
library_name = 'correlator'
tb_name = 'tb_correlator'
vhdl_generics = {}
vhdl_generics['g_nof_inputs'] = 8
other=None
tc = test_case.Testcase(tb_name.upper()+' - ', '')
io = node_io.NodeIO(tc.nodeImages, 'sim')
io.simIO.make_do_file(os.environ['RADIOHDL'], c_do_file_name, library_name, tb_name, vhdl_generics, other)
# Generate input data
correlator_snk_in_arr_complex = gen_complex_inputs(NOF_INPUTS, NOF_CHANNELS, COMPLEX_WIDTH)
# Convert to int and generate HEX files from input data
correlator_snk_in_arr_int = complex_to_int(correlator_snk_in_arr_complex)
gen_correlator_snk_in_arr_hex(correlator_snk_in_arr_int) # First setup the ModelSim environment for unb1
# cmd = '/home/schuur/SVN/RadioHDL/trunk/tools/modelsim/set_modelsim'
# args = 'unb1'
# subprocess.call([cmd, args])
# Calculate correlator output from input lists # Start modelsim in a separate terminal
correlator_src_out_arr_ref = correlate(correlator_snk_in_arr_int, INTEGRATION_PERIOD) # ms_cmd = ["konsole", "-e", "/home/software/Mentor/6.6c/modeltech/linux_x86_64/vsim", "-c", "-do", c_do_file_name]
# ms = subprocess.Popen(ms_cmd, shell=False, close_fds=True, preexec_fn=os.setsid)
ms_cmd = ["konsole", "-e", "run_modelsim", "unb1", "-do", c_do_file_name]
# (run TB)
if USE_NEW_REC_FILE==True:
# Wait for the correlator output
print 'Waiting for %s to be modified...' %REC_FILE
now = time.time()
do_until_gt(file_mod_time, now, s_timeout=900, filename=REC_FILE)
print '%s modified. Waiting for %d lines to be written...' %(REC_FILE, NOF_LINES_IN_REC_FILE_TO_VERIFY)
do_until_gt(file_linecount, NOF_LINES_IN_REC_FILE_TO_VERIFY, s_timeout=900, filename=REC_FILE)
else:
print 'Using old .rec file'
# Read test bench output file written by dp_stream_rec_play.vhd
correlator_src_out_arr = rec_file_to_complex(REC_FILE, complex_width=CORRELATOR_OUTPUT_COMPLEX_WIDTH)
# Compare modeloutput to VHDL output
compare_correlator_src_out_arr(correlator_src_out_arr, correlator_src_out_arr_ref)
if PLOT_CORR_OUT_PHASES==True:
plot_phase_shifts(correlator_src_out_arr[0])
plot_phase_shifts(correlator_src_out_arr_ref[0]) if __name__ == '__main__':
# Generate input data
correlator_snk_in_arr_complex = gen_complex_inputs(NOF_INPUTS, NOF_CHANNELS, COMPLEX_WIDTH)
# Convert to int and generate HEX files from input data
correlator_snk_in_arr_int = complex_to_int(correlator_snk_in_arr_complex, NOF_INPUTS, NOF_CHANNELS)
gen_correlator_snk_in_arr_hex(correlator_snk_in_arr_int)
# Calculate correlator output from input lists
correlator_src_out_arr_ref = correlate(correlator_snk_in_arr_int, INTEGRATION_PERIOD)
# (run TB)
#run_tb()
if USE_NEW_REC_FILE==True:
# Wait for the correlator output
print 'Waiting for %s to be modified...' %REC_FILE
now = time.time()
do_until_gt(file_mod_time, now, s_timeout=900, filename=REC_FILE)
print '%s modified. Waiting for %d lines to be written...' %(REC_FILE, NOF_LINES_IN_REC_FILE_TO_VERIFY)
do_until_gt(file_linecount, NOF_LINES_IN_REC_FILE_TO_VERIFY, s_timeout=900, filename=REC_FILE)
else:
print 'Using old .rec file'
# Read test bench output file written by dp_stream_rec_play.vhd
correlator_src_out_arr = rec_file_to_complex(REC_FILE, complex_width=CORRELATOR_OUTPUT_COMPLEX_WIDTH)
# Compare modeloutput to VHDL output
compare_correlator_src_out_arr(correlator_src_out_arr, correlator_src_out_arr_ref)
if PLOT_CORR_OUT_PHASES==True:
plot_phase_shifts(correlator_src_out_arr[0], NOF_VISIBILITIES)
plot_phase_shifts(correlator_src_out_arr_ref[0], NOF_VISIBILITIES)
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment