Add WaveformGen class to support waveform data generation.

Add APIs to generate data for various waveforms in the
genalyzer.py file
Update setup.py file
Add basic test support
Update test_genalyzer.py test script

Signed-off-by: SGudla <[email protected]>

bindings/python/genalyzer.py

@@ -20,6 +20,8 @@
 '''
 
 from dataclasses import dataclass, field
+import numpy as np
+from scipy import signal
 from typing import List
 from ctypes import (
     c_char,
@@ -872,4 +874,162 @@ def config_code_format(
     :param c: GNConfig object
     """
     code_format = c_uint(code_format)
-    _gn_config_set_code_format(code_format, byref(c._struct))
+    _gn_config_set_code_format(code_format, byref(c._struct))
+
+
+class WaveformGen:
+    """Waveform data generation for transmit devices"""
+
+    _tone_phase = [0, 0, 0]  # tone phase
+    _noise = 0.0
+
+    def __init__(self, npts: int, freq: int, code_fmt: int, res: int, v_ref_n: float, v_ref_p: float, v_min: float,
+                 v_max: float):
+        """Constructor for WaveformGen class.
+
+        :param npts: number of points required per waveform cycle
+        :param freq: output frequency required in hz
+        :param code_fmt: code format to get data in
+            0: for binary offset
+            1: for 2's complement
+        :param res: code resolution
+        :param v_ref_n: negative reference voltage
+            Can be zero(0) for unipolar device
+        :param v_ref_p: positive reference voltage
+        :param v_min: minimum required voltage to generate
+            Must be in the accepted reference voltage range
+        :param v_max: maximum required voltage to generate
+            Must be in the accepted reference voltage range
+        """
+
+        self._data = []
+        self.npts = npts
+        self.freq = freq
+        self.code_fmt = code_fmt
+        self.res = res
+        self._v_ref_n = v_ref_n
+        self._v_ref_p = v_ref_p
+        self.v_min = v_min
+        self.v_max = v_max
+
+    @property
+    def v_min(self):
+        """v_min: Lower required voltage limit"""
+        return self._v_min
+
+    @v_min.setter
+    def v_min(self, value):
+        """v_min: Set Lower required voltage limit"""
+        if value < self._v_ref_n:
+            raise Exception("required lower voltage cannot be less than lower voltage reference")
+        self._v_min = value
+
+    @property
+    def v_max(self):
+        """v_max: Upper required voltage limit"""
+        return self._v_max
+
+    @v_max.setter
+    def v_max(self, value):
+        """v_max: Set Upper required voltage limit"""
+        if value > self._v_ref_p:
+            raise Exception("required upper voltage cannot be greater than upper voltage reference")
+        self._v_max = value
+
+    def __prepare_waveform_gen(self):
+        self.__samp_freq = self.npts * self.freq
+        self.__maxcode = pow(2, self.res) - 1
+        self.__v_fsr = self._v_ref_p - self._v_ref_n
+        self.__vp_p = self.v_max - self.v_min
+        self.__mp_fsr = (self._v_ref_p + self._v_ref_n) / 2
+        self.__mp_vreq = (self.v_max + self.v_min) / 2
+        self.__offset = int(((self.__mp_vreq - self.__mp_fsr) / self.__v_fsr) * self.__maxcode)
+        self.__v_bias = ((abs(self._v_ref_n) + self.__mp_vreq) * 2) / self.__v_fsr
+        self.__amplitude = (2 ** (self.res - 1)) - 1
+
+        self.__tone_freq = [self.freq / self.npts] * 3
+        self.__tone_ampl = [self.__vp_p / 2] * 2
+
+        # Period
+        self.__ts = 1 / float(self.__samp_freq)
+        # Time array
+        self.__t = np.arange(0, self.npts * self.__ts, self.__ts)
+
+    def __gen_sine_cosine(self, tone_type):
+        self.__prepare_waveform_gen()
+
+        # Generate real tone
+        c = config_gen_tone(tone_type, self.npts, self.freq, 3, self.__tone_freq, self.__tone_ampl, self._tone_phase)
+        wf = gen_real_tone(c)
+
+        # Configure the waveform quantization
+        config_quantize(self.npts, self.__v_fsr, self.res, self._noise, c)
+
+        # Configure code data format
+        config_code_format(self.code_fmt, c)
+
+        # Get the quantized waveform
+        qwf = quantize(wf, c)
+        self._data = [x + self.__offset for x in qwf]  # Add the offset calculated
+
+        # Free config object
+        config_free(c)
+
+        return self._data
+
+    def __gen_other_waveforms(self):
+        self._data *= (self.__vp_p / self.__v_fsr)
+        self._data += self.__v_bias
+        self._data *= self.__amplitude
+        self._data = np.uint32(self._data)
+
+        if self.code_fmt:
+            self._data = np.bitwise_xor(2 ** (self.res - 1), self._data)
+
+        return self._data
+
+    def __del__(self):
+        self._data = []
+
+    def gen_sine_wave(self):
+        """Generate sine wave data
+
+        :return: waveform as list of ints
+        """
+        return self.__gen_sine_cosine(1)
+
+    def gen_cosine_wave(self):
+        """Generate cosine wave data
+
+        :return: waveform as list of ints
+        """
+        return self.__gen_sine_cosine(0)
+
+    def gen_triangular_wave(self):
+        """Generate triangular wave data
+
+        :return: waveform as list of ints
+        """
+        self.__prepare_waveform_gen()
+        self._data = signal.sawtooth(2 * np.pi * self.freq * self.__t, 0.5)
+        return self.__gen_other_waveforms()
+
+    def gen_square_wave(self):
+        """Generate square wave data
+
+        :return: waveform as list of ints
+        """
+        self.__prepare_waveform_gen()
+        self._data = signal.square(2 * np.pi * self.freq * self.__t, 0.5)
+        return self.__gen_other_waveforms()
+
+    def gen_pwm_wave(self, duty_cycle):
+        """Generate pwm wave data
+        :param duty_cycle: Duty cycle required
+            Must be in between 0 and 1 with a precision of unto two decimals.
+            e.g. 0.25 for 25% duty-cycle
+        :return: waveform as list of ints
+        """
+        self.__prepare_waveform_gen()
+        self._data = signal.square(2 * np.pi * self.freq * self.__t, duty_cycle)
+        return self.__gen_other_waveforms()

bindings/python/setup.py.cmakein

@@ -85,6 +85,7 @@ config.update(
         py_modules=["genalyzer","genalyzer_advanced"],
         packages=find_packages(exclude=["test*"]),
         python_requires=">=3.6",
+        install_requires=['scipy>=1.11.2', 'numpy>=1.22.4'],
         cmdclass={"install": InstallWrapper},
         classifiers=[
             "Programming Language :: Python :: 3",

bindings/python/tests/test_genalyzer.py

@@ -18,6 +18,9 @@
 test_fft_tone_files = [
     f for f in glob.glob(os.path.join(loc, test_dir, "test_fft_tone_*.json"))
 ]
+test_gen_wave_data = [
+    f for f in glob.glob(os.path.join(loc, test_dir, "test_gen_wave_data_*.json"))
+]
 
 @pytest.mark.parametrize("filename", test_gen_real_tone_files)
 def test_gen_real_tone(filename):
@@ -97,7 +100,7 @@ def test_get_fa_results(filename):
     with open(filename) as f:
         data = json.load(f)
         if data['num_tones'] == 1:
-            freq_list = [data['freq']]            
+            freq_list = [data['freq']]
         else:
             freq_list = data['freq']
 
@@ -119,7 +122,7 @@ def test_get_fa_single_result(filename):
     with open(filename) as f:
         data = json.load(f)
         if data['num_tones'] == 1:
-            freq_list = [data['freq']]            
+            freq_list = [data['freq']]
         else:
             freq_list = data['freq']
 
@@ -157,4 +160,21 @@ def test_get_wfa_results(filename):
         c = genalyzer.config_quantize(data['npts'], data['fsr'], data['qres'], data['qnoise'])
         wfa_results = genalyzer.get_wfa_results(qwf, c)
         genalyzer.config_free(c)
-        assert bool(wfa_results), "the dict is non empty"
+        assert bool(wfa_results), "the dict is non empty"
+
+@pytest.mark.parametrize("filename", test_gen_wave_data)
+def test_gen_wf_data(filename):
+    with open(filename) as f:
+        data = json.load(f)
+        wavegen = genalyzer.WaveformGen(data['npts'], data['freq'], data['code_fmt'], data['res'], data['v_ref_n'],
+                                        data['v_ref_p'], data['v_min'], data['v_max'])
+        wf_sine = wavegen.gen_sine_wave()
+        wf_cosine = wavegen.gen_cosine_wave()
+        wf_tri = wavegen.gen_triangular_wave()
+        wf_square = wavegen.gen_square_wave()
+        wf_pwm = wavegen.gen_pwm_wave(data['duty_cyc'])
+        assert len(wf_sine) != 0, "the list is empty"
+        assert len(wf_cosine) != 0, "the list is empty"
+        assert len(wf_tri) != 0, "the list is empty"
+        assert len(wf_square) != 0, "the list is empty"
+        assert len(wf_pwm) != 0, "the list is empty"