From 339e134013e0da21d00817b5510247ce8272fd46 Mon Sep 17 00:00:00 2001 From: Adrian Stanea Date: Mon, 19 Aug 2024 17:26:05 +0300 Subject: [PATCH] tests: refactor duplicate code - Include helpers.py for common functions Signed-off-by: Adrian Stanea --- tests/analog_functions.py | 193 +++++----------------------------- tests/bug_checks_functions.py | 44 +------- tests/digital_functions.py | 94 ++--------------- tests/helpers.py | 59 +++++++++++ tests/m2k_digital_test.py | 3 +- tests/reset_def_values.py | 14 +++ 6 files changed, 114 insertions(+), 293 deletions(-) create mode 100644 tests/helpers.py diff --git a/tests/analog_functions.py b/tests/analog_functions.py index e5f230f5..7ce4e43d 100644 --- a/tests/analog_functions.py +++ b/tests/analog_functions.py @@ -13,6 +13,7 @@ import random import sys import reset_def_values as reset +from helpers import get_result_files, save_data_to_csv, plot_to_file from open_context import ctx_timeout, ctx from create_files import results_file, results_dir, csv @@ -133,16 +134,8 @@ def test_amplitude(out_data, ref_data, n, ain, aout, channel, trig): # corr_amplitude_min -- correlation coefficient between the vector that holds the minimum amplitude values in the # input signal and the vector that holds the maximum amplitude values in the reference signal - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) + reset.analog_in(ain) reset.analog_out(aout) @@ -204,7 +197,7 @@ def test_amplitude(out_data, ref_data, n, ain, aout, channel, trig): data_string.append("Minimum amplitude computed:") data_string.append(str(min_in)) if gen_reports: - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) aout.stop(channel) return corr_amplitude_max, corr_amplitude_min @@ -229,16 +222,7 @@ def test_shape(channel, out_data, ref_data, ain, aout, trig, ch_ratio, shapename # phase_diff_vect-- vector that holds the phase difference between the input data and the reference data for each # signal shape - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) reset.analog_in(ain) reset.analog_out(aout) @@ -284,7 +268,7 @@ def test_shape(channel, out_data, ref_data, ain, aout, trig, ch_ratio, shapename "Correlation coefficient between " + shapename[i] + "signal and its reference:" + str(corr_shape)) data_string.append("Phase difference between " + shapename[i] + "signal and its reference:" + str(phase_diff)) if gen_reports: - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) aout.stop(channel) return corr_shape_vect, phase_diff_vect @@ -304,16 +288,8 @@ def phase_diff_ch0_ch1(aout, ain, trig): # Returns: # phase_diff_between_channels-- the phase difference between channels in degrees # - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + + file_name, dir_name, csv_path = get_result_files(gen_reports) reset.analog_in(ain) reset.analog_out(aout) @@ -359,7 +335,7 @@ def phase_diff_ch0_ch1(aout, ain, trig): phasediff_csv['Ch1, ADCsr=' + str(sr)] = input_data[libm2k.ANALOG_IN_CHANNEL_2] if gen_reports: channel = 0 - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) save_data_to_csv(phasediff_csv, csv_path + 'ph_diff_channels.csv') aout.stop() @@ -380,16 +356,7 @@ def test_analog_trigger(channel, trig, aout, ain): # trig_test -- Vector that holds 1 for each trigger condition fulfilled and 0 otherwise # condition_name-- Vector that holds names of the trigger conditions - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) reset.analog_in(ain) reset.analog_out(aout) @@ -531,7 +498,7 @@ def test_analog_trigger(channel, trig, aout, ain): data_string.append("level set: " + str(high)) data_string.append("\nlevel read: " + str(input_data[delay])) if gen_reports: - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) aout.stop(channel) return trig_test, condition_name @@ -552,16 +519,7 @@ def test_offset(out_data, n, ain, aout, trig, channel): # Returns # corr_offset -- Correlation coefficient between the computed offset vector and the defined offset vector - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) reset.analog_in(ain) reset.analog_out(aout) @@ -604,7 +562,7 @@ def test_offset(out_data, n, ain, aout, trig, channel): data_string.append(str(in_offset)) if gen_reports: - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) corr_offset, _ = pearsonr(offset, in_offset) # compare the original offset vector with the average values obtained aout.stop(channel) @@ -623,16 +581,7 @@ def test_voltmeter_functionality(channel, ain, aout, ctx): # Returns: # voltmeter_ -- Vector thet holds 1 if the read voltage is in specified range and 0 otherwise - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) reset.analog_in(ain) reset.analog_out(aout) @@ -666,7 +615,7 @@ def test_voltmeter_functionality(channel, ain, aout, ctx): else: voltmeter_ = np.append(voltmeter_, 0) # the voltage is out of range if gen_reports: - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) aout.stop(channel) return voltmeter_ @@ -699,16 +648,7 @@ def cyclic_buffer_test(aout, ain, channel, trig): # Returns: # cyclic_false -- Must be 1 if a single buffer was sent and succesfully recieved, 0 otherwise - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) dac_sr = 75000 adc_sr = 10000 min_periods = 3 @@ -754,16 +694,7 @@ def noncyclic_buffer_test(aout, ain, channel, trig, ctx): # Returns: # cyclic_false -- Must be 1 if a single buffer was sent and succesfully recieved, 0 otherwise - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) reset.analog_in(ain) reset.analog_out(aout) @@ -855,16 +786,7 @@ def compute_frequency(channel, ain, aout, trig): # ofreqs-- Vector that holds the frequencies of the output buffers # ifreqs-- Vector that holds the frequencies of the input buffers - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) test_name = "freq" data_string = [] @@ -924,7 +846,7 @@ def compute_frequency(channel, ain, aout, trig): data_string.append("Out signal frequency:" + str(out_freq)) data_string.append("In singal frequency:" + str(in_freq)) if gen_reports: - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) return ofreqs, ifreqs @@ -987,16 +909,7 @@ def test_oversampling_ratio(channel, ain, aout, trig): # test_osr -- Must be 1 if the computed oversampling ratio is equal with the set oversampling ratio # and 0 otherwise - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) reset.analog_in(ain) reset.analog_out(aout) @@ -1044,7 +957,7 @@ def test_oversampling_ratio(channel, ain, aout, trig): data_string.append("Oversampling ratios computed: \n" + str(verify_osr)) if gen_reports: - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) test_osr = 1 for i in range(len(osr)): if osr[i] != verify_osr[i]: @@ -1053,44 +966,6 @@ def test_oversampling_ratio(channel, ain, aout, trig): return test_osr -def plot_to_file(title, data, dir_name, filename, xlabel=None, ylabel=None, data1=None, data_marked=None): - # Saves the plots in a separate folder - # Arguments: - # title -- Title of the plot - # data -- Data to be plotted - # filename -- Name of the file with the plot - # Keyword Arguments: - # xlabel -- Label of x-Axis (default: {None}) - # ylabel -- Label of y-Axis(default: {None}) - # data1 -- Data that should be plotted on the same plot(default: {None}) - # data_marked -- Data that represents specific points on the plot(default: {None}) - # plot the signals in a separate folder - plt.title(title) - if xlabel is not None: # if xlabel and ylabel are not specified there will be default values - plt.xlabel(xlabel) - else: - plt.xlabel('Samples') - if ylabel is not None: - plt.ylabel(ylabel) - else: - plt.ylabel('Voltage [V]') - plt.grid(visible=True) - plt.plot(data) # if a second set of data must be printed (for ch0 and ch1 phase difference in this case) - if data1 is not None: - plt.plot(data1) - if data_marked is not None: - plt.plot(data_marked, data[data_marked], 'xr') - plt.savefig(dir_name + "/" + filename) - plt.close() - return - - -def save_data_to_csv(csv_vals, csv_file): - df = DataFrame(csv_vals) - df.to_csv(csv_file) - return - - def channels_diff_in_samples(trig, channel, aout, ain): # Find if there is a sample delay between channels for the same signal and trigger # Arguments: @@ -1107,16 +982,7 @@ def channels_diff_in_samples(trig, channel, aout, ain): # dac_osr-- oversampling ratios set for Aout # freq -- frequency of the signals used for the test - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) reset.analog_in(ain) reset.analog_out(aout) # /len(samples_diff1[i]) reset.trigger(trig) @@ -1172,7 +1038,7 @@ def channels_diff_in_samples(trig, channel, aout, ain): data_string.append('Samples difference: ' + str(diff_osr)) if gen_reports: - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) save_data_to_csv(diff_in_samples0, csv_path + 'diffSamples_ch0_trigSrc_' + str(channel) + '.csv') save_data_to_csv(diff_in_samples1, csv_path + 'diffSamples_ch1_trigSrc_' + str(channel) + '.csv') aout.stop() @@ -1317,16 +1183,7 @@ def is_spike(data, peak, threshold = 0.25): return percentage_dif > threshold def test_buffer_transition_glitch(channel, ain, aout, trig, waveform, amplitude=1): - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) BUFFER_SIZE = 5_00_000 @@ -1401,7 +1258,7 @@ def test_buffer_transition_glitch(channel, ain, aout, trig, waveform, amplitude= "Number of glitch peaks found in " + waveform + " signal :" + str(num_peaks)) if gen_reports: - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) plot_to_file(f'Buffer Glitch , channel{channel}', data, dir_name, diff --git a/tests/bug_checks_functions.py b/tests/bug_checks_functions.py index 19e08ef4..1bf22180 100644 --- a/tests/bug_checks_functions.py +++ b/tests/bug_checks_functions.py @@ -2,6 +2,7 @@ import matplotlib.pyplot as plt import libm2k +from helpers import get_result_files, plot_to_file from open_context import ctx_timeout, ctx, ain, aout, trig gen_reports = True @@ -9,16 +10,8 @@ def test_dac_artifact(ain, aout, trig, chn): # check for DAC artifacts at slow sample rates - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) + buffer_size = 4000 sampling_frequency_in = 1000000 sampling_frequency_out = 750000 @@ -63,33 +56,4 @@ def generate_clock_signal(): buffer.append(1) for i in range(256): buffer.append(0) - return buffer - - -def plot_to_file(title, data, dir_name, filename, xlabel=None, ylabel=None, data1=None): - # Saves the plots in a separate folder - # Arguments: - # title -- Title of the plot - # data -- Data to be plotted - # filename -- Name of the file with the plot - # Keyword Arguments: - # xlabel -- Label of x-Axis (default: {None}) - # ylabel -- Label of y-Axis(default: {None}) - # data1 -- Data that should be plotted on the same plot(default: {None}) - # plot the signals in a separate folder - plt.title(title) - if xlabel is not None: # if xlabel and ylabel are not specified there will be default values - plt.xlabel(xlabel) - else: - plt.xlabel('Samples') - if ylabel is not None: - plt.ylabel(ylabel) - else: - plt.ylabel('Voltage [V]') - plt.grid(visible=True) - plt.plot(data) # if a second set of data must be printed (for ch0 and ch1 phase difference in this case) - if data1 is not None: - plt.plot(data1) - plt.savefig(dir_name + "/" + filename) - plt.close() - return + return buffer \ No newline at end of file diff --git a/tests/digital_functions.py b/tests/digital_functions.py index 8f2cfcf1..246d6580 100644 --- a/tests/digital_functions.py +++ b/tests/digital_functions.py @@ -3,48 +3,18 @@ import libm2k from multiprocessing.pool import ThreadPool import threading -from pandas import DataFrame import random import time from open_context import ctx +from helpers import get_result_files, save_data_to_csv, plot_to_file +import reset_def_values as reset + from scipy.signal import find_peaks sample = random.randint(0, 255) gen_reports = True -def result_files(gen_reports): - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] - dir_name = [] - csv_path = [] - - return file, dir_name, csv_path - - -def dig_reset(dig): - # Reset digital object - # Arguments: - # dig -- Digital object - - dig.setSampleRateIn(10000) - dig.setSampleRateOut(10000) - dig.setCyclic(True) - dig.setDirection(0b1111111111111111) - for i in range(16): - dig.setOutputMode(i, 1) - dig.enableAllOut(True) - return - - def set_digital_trigger(dig): # Set the digital trigger # Arguments: @@ -59,7 +29,6 @@ def set_digital_trigger(dig): def check_digital_channels_state(dig, channel): - dig.reset() # enable channel under test dig.setDirection(channel, libm2k.DIO_OUTPUT) @@ -142,8 +111,9 @@ def get_data_to_check_trig_condition(dig, channel, i, buff): def check_digital_trigger(channel, dig, d_trig): - file_name, dir_name, csv_path = result_files(gen_reports) - dig_reset(dig) + file_name, dir_name, csv_path = get_result_files(gen_reports) + + reset.digital(dig) delay = 1 condition = [libm2k.RISING_EDGE_DIGITAL, libm2k.FALLING_EDGE_DIGITAL, libm2k.LOW_LEVEL_DIGITAL, @@ -249,7 +219,8 @@ def task1(nb_samples, dig): def test_digital_cyclic_buffer(dig, d_trig, channel): - file, dir_name, csv_path = result_files(gen_reports) + file, dir_name, csv_path = get_result_files(gen_reports) + dig.setDirection(channel, libm2k.DIO_OUTPUT) dig.enableChannel(channel, True) d_trig.setDigitalCondition(channel, libm2k.LOW_LEVEL_DIGITAL) @@ -286,41 +257,6 @@ def test_digital_cyclic_buffer(dig, d_trig, channel): return passed -def plot_to_file(title, data, dir_name, filename, xlabel=None, ylabel=None, data1=None): - # Saves the plots in a separate folder - # Arguments: - # title -- Title of the plot\n - # data -- Data to be plotted\n - # filename -- Name of the file with the plot\n - # Keyword Arguments: - # xlabel -- Label of x-Axis (default: {None}) - # ylabel -- Label of y-Axis(default: {None}) - # data1 -- Data that should be plotted on the same plot(default: {None}) - - # plot the signals in a separate folder - plt.title(title) - if xlabel is not None: # if xlabel and ylabel are not specified there will be default values - plt.xlabel(xlabel) - else: - plt.xlabel('Samples') - if ylabel is not None: - plt.ylabel(ylabel) - else: - plt.ylabel('Voltage [V]') - plt.grid(visible=True) - plt.plot(data) # if a second set of data must be printed (for ch0 and ch1 phase difference in this case) - if data1 is not None: - plt.plot(data1) - plt.savefig(dir_name + "/" + filename) - plt.close() - return - - -def save_data_to_csv(csv_vals, csv_file): - df = DataFrame(csv_vals) - df.to_csv(csv_file) - return - def test_kernel_buffers(dig, nb_kernel_buffers): error = False dig.reset() @@ -351,16 +287,7 @@ def count_edges(data, threshold = 0.5): def test_pattern_generator_pulse(dig, d_trig, channel): - if gen_reports: - from create_files import results_file, results_dir, csv, open_files_and_dirs - if results_file is None: - file, dir_name, csv_path = open_files_and_dirs() - else: - file = results_file - dir_name = results_dir - csv_path = csv - else: - file = [] + file_name, dir_name, csv_path = get_result_files(gen_reports) timeout = 15_000 # in milliseconds delay = 8192 @@ -371,7 +298,6 @@ def test_pattern_generator_pulse(dig, d_trig, channel): test_name = "pattern_generator_pulse" data_string = [] - file_name, dir_name, csv_path = result_files(gen_reports) dig.reset() ctx.setTimeout(timeout) @@ -414,7 +340,7 @@ def test_pattern_generator_pulse(dig, d_trig, channel): "Expected: " + str(expected_num_edges) + " , found: " + str(actual_num_edges)) if gen_reports: - write_file(file, test_name, channel, data_string) + write_file(file_name, test_name, channel, data_string) plot_to_file("Pattern generator on ch" + str(channel), crnt_chn_dio_data, dir_name, "digital_pattern_generator_ch" + str(channel) + ".png") diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 00000000..ea52f605 --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,59 @@ + +from pandas import DataFrame +import matplotlib.pyplot as plt + + + +def get_result_files(gen_reports): + if gen_reports: + from create_files import results_file, results_dir, csv, open_files_and_dirs + if results_file is None: + file, dir_name, csv_path = open_files_and_dirs() + else: + file = results_file + dir_name = results_dir + csv_path = csv + else: + file = [] + dir_name = [] + csv_path = [] + + return file, dir_name, csv_path + + +def save_data_to_csv(csv_vals, csv_file): + df = DataFrame(csv_vals) + df.to_csv(csv_file) + return + + +def plot_to_file(title, data, dir_name, filename, xlabel=None, ylabel=None, data1=None, data_marked=None): + # Saves the plots in a separate folder + # Arguments: + # title -- Title of the plot + # data -- Data to be plotted + # filename -- Name of the file with the plot + # Keyword Arguments: + # xlabel -- Label of x-Axis (default: {None}) + # ylabel -- Label of y-Axis(default: {None}) + # data1 -- Data that should be plotted on the same plot(default: {None}) + # data_marked -- Data that represents specific points on the plot(default: {None}) + # plot the signals in a separate folder + plt.title(title) + if xlabel is not None: # if xlabel and ylabel are not specified there will be default values + plt.xlabel(xlabel) + else: + plt.xlabel('Samples') + if ylabel is not None: + plt.ylabel(ylabel) + else: + plt.ylabel('Voltage [V]') + plt.grid(visible=True) + plt.plot(data) # if a second set of data must be printed (for ch0 and ch1 phase difference in this case) + if data1 is not None: + plt.plot(data1) + if data_marked is not None: + plt.plot(data_marked, data[data_marked], 'xr') + plt.savefig(dir_name + "/" + filename) + plt.close() + return \ No newline at end of file diff --git a/tests/m2k_digital_test.py b/tests/m2k_digital_test.py index 20e281c9..f168a621 100644 --- a/tests/m2k_digital_test.py +++ b/tests/m2k_digital_test.py @@ -1,8 +1,9 @@ import unittest import libm2k -from digital_functions import dig_reset, set_digital_trigger, check_digital_channels_state, check_digital_output, \ +from digital_functions import set_digital_trigger, check_digital_channels_state, check_digital_output, \ check_digital_trigger, check_open_drain_mode, test_kernel_buffers, test_pattern_generator_pulse from digital_functions import test_digital_cyclic_buffer +import reset_def_values as reset from open_context import ctx, dig, d_trig import logger from repeat_test import repeat diff --git a/tests/reset_def_values.py b/tests/reset_def_values.py index 28ceadce..d0a22bb0 100644 --- a/tests/reset_def_values.py +++ b/tests/reset_def_values.py @@ -56,3 +56,17 @@ def trigger(trig): trig.setAnalogDelay(0) return + + +def digital(dig: libm2k.M2kDigital): + # Sets default values for digital parameters + # Arguments: + # dig -- Digital object + + dig.setSampleRateIn(10000) + dig.setSampleRateOut(10000) + dig.setCyclic(True) + dig.setDirection(0b1111111111111111) + for i in range(16): + dig.setOutputMode(i, 1) + dig.enableAllOut(True)