Add fancy write_source_waveform_unique

CHANGELOG.md

@@ -55,6 +55,7 @@ All notable changes to this project will be documented in this file.
 * ### NI-Digital Pattern Driver
     * #### Added
         * `fetch_capture_waveform()` - returns dictionary { site: data, site: data, ... }
+        * `write_source_waveform_site_unique()` - takes waveform_name and dictionary { site: data, site: data, ... }
         * `pins` is now a valid repeated capability
     * #### Changed
         * Fix get/set properties - [#1062](https://github.com/ni/nimi-python/issues/1062)

docs/nidigital/class.rst

@@ -3370,6 +3370,36 @@ write_source_waveform_data_from_file_tdms
 
             :type waveform_file_path: str
 
+write_source_waveform_site_unique
+---------------------------------
+
+    .. py:currentmodule:: nidigital.Session
+
+    .. py:method:: write_source_waveform_site_unique(waveform_name, waveform_data)
+
+            TBD
+
+
+
+
+
+            :param waveform_name:
+
+
+
+
+
+            :type waveform_name: str
+            :param waveform_data:
+
+
+                Dictionary where each key is the site number and the value is array.array of unsigned int
+
+
+
+
+            :type waveform_data: { site: data, site: data, ... }
+
 write_static
 ------------
 

generated/nidigital/_library.py

@@ -137,6 +137,7 @@ def __init__(self, ctypes_library):
         self.niDigital_WriteSequencerRegister_cfunc = None
         self.niDigital_WriteSourceWaveformBroadcastU32_cfunc = None
         self.niDigital_WriteSourceWaveformDataFromFileTDMS_cfunc = None
+        self.niDigital_WriteSourceWaveformSiteUniqueU32_cfunc = None
         self.niDigital_WriteStatic_cfunc = None
         self.niDigital_close_cfunc = None
         self.niDigital_error_message_cfunc = None
@@ -1095,6 +1096,14 @@ def niDigital_WriteSourceWaveformDataFromFileTDMS(self, vi, waveform_name, wavef
                 self.niDigital_WriteSourceWaveformDataFromFileTDMS_cfunc.restype = ViStatus  # noqa: F405
         return self.niDigital_WriteSourceWaveformDataFromFileTDMS_cfunc(vi, waveform_name, waveform_file_path)
 
+    def niDigital_WriteSourceWaveformSiteUniqueU32(self, vi, site_list, waveform_name, num_waveforms, samples_per_waveform, waveform_data):  # noqa: N802
+        with self._func_lock:
+            if self.niDigital_WriteSourceWaveformSiteUniqueU32_cfunc is None:
+                self.niDigital_WriteSourceWaveformSiteUniqueU32_cfunc = self._library.niDigital_WriteSourceWaveformSiteUniqueU32
+                self.niDigital_WriteSourceWaveformSiteUniqueU32_cfunc.argtypes = [ViSession, ctypes.POINTER(ViChar), ctypes.POINTER(ViChar), ViInt32, ViInt32, ctypes.POINTER(ViUInt32)]  # noqa: F405
+                self.niDigital_WriteSourceWaveformSiteUniqueU32_cfunc.restype = ViStatus  # noqa: F405
+        return self.niDigital_WriteSourceWaveformSiteUniqueU32_cfunc(vi, site_list, waveform_name, num_waveforms, samples_per_waveform, waveform_data)
+
     def niDigital_WriteStatic(self, vi, channel_list, state):  # noqa: N802
         with self._func_lock:
             if self.niDigital_WriteStatic_cfunc is None:

generated/nidigital/session.py

@@ -2845,6 +2845,47 @@ def self_test(self):
             raise errors.SelfTestError(code, msg)
         return None
 
+    @ivi_synchronized
+    def write_source_waveform_site_unique(self, waveform_name, waveform_data):
+        '''write_source_waveform_site_unique
+
+        TBD
+
+        Args:
+            waveform_name (str):
+
+            waveform_data ({ site: data, site: data, ... }): Dictionary where each key is the site number and the value is array.array of unsigned int
+
+        '''
+        site_list = []
+        # We assume all the entries are the same length (we'll check later) to make the array the correct size
+        # Get an entry from the dictionary from https://stackoverflow.com/questions/30362391/how-do-you-find-the-first-key-in-a-dictionary
+        if len(waveform_data) == 0:
+            actual_samples_per_waveform = 0
+        else:
+            actual_samples_per_waveform = len(waveform_data[next(iter(waveform_data))])
+        data = array.array('L', [0] * (len(waveform_data) * actual_samples_per_waveform))
+        mv = memoryview(data)
+
+        i = 0
+        for site in waveform_data:
+            if len(waveform_data[site]) != actual_samples_per_waveform:
+                raise ValueError('Mismatched length of waveforms. All must be the same length.')
+            if waveform_data[site].typecode != 'L':
+                raise ValueError('Wrong array element type. Must be unsigned 32 bit int ("L"), was {}'.format(waveform_data[site].typecode))
+
+            site_list.append('site' + str(site))
+
+            start = i * actual_samples_per_waveform
+            end = start + actual_samples_per_waveform
+            mv[start:end] = waveform_data[site]
+
+            i += 1
+
+        site_list_str = ','.join(site_list)
+
+        self._write_source_waveform_site_unique_u32(site_list_str, waveform_name, len(waveform_data), actual_samples_per_waveform, data)
+
     @ivi_synchronized
     def fetch_history_ram_cycle_information(self, site, sample_index):
         r'''fetch_history_ram_cycle_information
@@ -3626,6 +3667,35 @@ def write_source_waveform_data_from_file_tdms(self, waveform_name, waveform_file
         errors.handle_error(self, error_code, ignore_warnings=False, is_error_handling=False)
         return
 
+    @ivi_synchronized
+    def _write_source_waveform_site_unique_u32(self, site_list, waveform_name, num_waveforms, samples_per_waveform, waveform_data):
+        r'''_write_source_waveform_site_unique_u32
+
+        TBD
+
+        Args:
+            site_list (str):
+
+            waveform_name (str):
+
+            num_waveforms (int):
+
+            samples_per_waveform (int):
+
+            waveform_data (array.array("L")):
+
+        '''
+        vi_ctype = _visatype.ViSession(self._vi)  # case S110
+        site_list_ctype = ctypes.create_string_buffer(site_list.encode(self._encoding))  # case C020
+        waveform_name_ctype = ctypes.create_string_buffer(waveform_name.encode(self._encoding))  # case C020
+        num_waveforms_ctype = _visatype.ViInt32(num_waveforms)  # case S150
+        samples_per_waveform_ctype = _visatype.ViInt32(samples_per_waveform)  # case S150
+        waveform_data_array = get_ctypes_and_array(value=waveform_data, array_type="L")  # case B550
+        waveform_data_ctype = get_ctypes_pointer_for_buffer(value=waveform_data_array, library_type=_visatype.ViUInt32)  # case B550
+        error_code = self._library.niDigital_WriteSourceWaveformSiteUniqueU32(vi_ctype, site_list_ctype, waveform_name_ctype, num_waveforms_ctype, samples_per_waveform_ctype, waveform_data_ctype)
+        errors.handle_error(self, error_code, ignore_warnings=False, is_error_handling=False)
+        return
+
     def _close(self):
         r'''_close
 

generated/nidigital/unit_tests/_mock_helper.py

@@ -307,6 +307,8 @@ def __init__(self):
         self._defaults['WriteSourceWaveformBroadcastU32']['return'] = 0
         self._defaults['WriteSourceWaveformDataFromFileTDMS'] = {}
         self._defaults['WriteSourceWaveformDataFromFileTDMS']['return'] = 0
+        self._defaults['WriteSourceWaveformSiteUniqueU32'] = {}
+        self._defaults['WriteSourceWaveformSiteUniqueU32']['return'] = 0
         self._defaults['WriteStatic'] = {}
         self._defaults['WriteStatic']['return'] = 0
         self._defaults['close'] = {}
@@ -1262,6 +1264,11 @@ def niDigital_WriteSourceWaveformDataFromFileTDMS(self, vi, waveform_name, wavef
             return self._defaults['WriteSourceWaveformDataFromFileTDMS']['return']
         return self._defaults['WriteSourceWaveformDataFromFileTDMS']['return']
 
+    def niDigital_WriteSourceWaveformSiteUniqueU32(self, vi, site_list, waveform_name, num_waveforms, samples_per_waveform, waveform_data):  # noqa: N802
+        if self._defaults['WriteSourceWaveformSiteUniqueU32']['return'] != 0:
+            return self._defaults['WriteSourceWaveformSiteUniqueU32']['return']
+        return self._defaults['WriteSourceWaveformSiteUniqueU32']['return']
+
     def niDigital_WriteStatic(self, vi, channel_list, state):  # noqa: N802
         if self._defaults['WriteStatic']['return'] != 0:
             return self._defaults['WriteStatic']['return']
@@ -1548,6 +1555,8 @@ def set_side_effects_and_return_values(self, mock_library):
         mock_library.niDigital_WriteSourceWaveformBroadcastU32.return_value = 0
         mock_library.niDigital_WriteSourceWaveformDataFromFileTDMS.side_effect = MockFunctionCallError("niDigital_WriteSourceWaveformDataFromFileTDMS")
         mock_library.niDigital_WriteSourceWaveformDataFromFileTDMS.return_value = 0
+        mock_library.niDigital_WriteSourceWaveformSiteUniqueU32.side_effect = MockFunctionCallError("niDigital_WriteSourceWaveformSiteUniqueU32")
+        mock_library.niDigital_WriteSourceWaveformSiteUniqueU32.return_value = 0
         mock_library.niDigital_WriteStatic.side_effect = MockFunctionCallError("niDigital_WriteStatic")
         mock_library.niDigital_WriteStatic.return_value = 0
         mock_library.niDigital_close.side_effect = MockFunctionCallError("niDigital_close")

src/nidigital/metadata/functions.py

@@ -3772,7 +3772,7 @@
         'returns': 'ViStatus'
     },
     'WriteSourceWaveformSiteUniqueU32': {
-        'codegen_method': 'no',
+        'codegen_method': 'private',
         'documentation': {
             'description': 'TBD'
         },
@@ -3805,10 +3805,7 @@
             {
                 'direction': 'in',
                 'name': 'waveformData',
-                'size': {
-                    'mechanism': 'custom',
-                    'value': None
-                },
+                'use_array': True,
                 'type': 'ViUInt32[]'
             }
         ],
@@ -3954,3 +3951,4 @@
         'returns': 'ViStatus'
     }
 }
+

src/nidigital/metadata/functions_addon.py

@@ -4,6 +4,44 @@
 functions_override_metadata = {
 }
 
+functions_additional_write_source_waveform_site_unique = {
+    'FancyWriteSourceWaveformSiteUnique': {
+        'python_name': 'write_source_waveform_site_unique',
+        'codegen_method': 'python-only',
+        'method_templates': [
+            {
+                'documentation_filename': 'default_method',
+                'method_python_name_suffix': '',
+                'session_filename': 'fancy_write_source_waveform_site_unique',
+            }
+        ],
+        'documentation': {
+            'description': 'TBD'
+        },
+        'parameters': [
+            {
+                'direction': 'in',
+                'name': 'vi',
+                'type': 'ViSession'
+            },
+            {
+                'direction': 'in',
+                'name': 'waveformName',
+                'type': 'ViConstString'
+            },
+            {
+                'direction': 'in',
+                'documentation': {
+                    'description': '\nDictionary where each key is the site number and the value is array.array of unsigned int\n'
+                },
+                'name': 'waveform_data',
+                'type': 'ViUInt32',  # This type is ignored since this function isn't code generated
+                'type_in_documentation': '{ site: data, site: data, ... }',
+            },
+        ],
+    },
+}
+
 functions_additional_get_pin_results_pin_information = {
     'FancyGetPinResultsPinInformation': {
         'python_name': 'get_pin_results_pin_information',

src/nidigital/system_tests/test_system_nidigital.py

@@ -68,6 +68,12 @@ def test_tdr_some_channels(multi_instrument_session):
     assert fetched_offsets == applied_offsets
 
 
+def test_write_source_waveform_site_unique(multi_instrument_session):
+    # Implementation of write_source_waveform_site_unique() uses memoryview of array.array, which is not supported in Python 2.x
+    if sys.version_info.major >= 3:
+        pass
+
+
 def test_get_pin_results_pin_information(multi_instrument_session):
     pass
 

src/nidigital/templates/session.py/fancy_write_source_waveform_site_unique.py.mako

@@ -0,0 +1,39 @@
+<%page args="f, config, method_template"/>\
+<%
+    import build.helper as helper
+    suffix = method_template['method_python_name_suffix']
+%>\
+    def ${f['python_name']}${suffix}(${helper.get_params_snippet(f, helper.ParameterUsageOptions.SESSION_METHOD_DECLARATION)}):
+        '''${f['python_name']}
+
+        ${helper.get_function_docstring(f, False, config, indent=8)}
+        '''
+        site_list = []
+        # We assume all the entries are the same length (we'll check later) to make the array the correct size
+        # Get an entry from the dictionary from https://stackoverflow.com/questions/30362391/how-do-you-find-the-first-key-in-a-dictionary
+        if len(waveform_data) == 0:
+            actual_samples_per_waveform = 0
+        else:
+            actual_samples_per_waveform = len(waveform_data[next(iter(waveform_data))])
+        data = array.array('L', [0] * (len(waveform_data) * actual_samples_per_waveform))
+        mv = memoryview(data)
+
+        i = 0
+        for site in waveform_data:
+            if len(waveform_data[site]) != actual_samples_per_waveform:
+                raise ValueError('Mismatched length of waveforms. All must be the same length.')
+            if waveform_data[site].typecode != 'L':
+                raise ValueError('Wrong array element type. Must be unsigned 32 bit int ("L"), was {}'.format(waveform_data[site].typecode))
+
+            site_list.append('site' + str(site))
+
+            start = i * actual_samples_per_waveform
+            end = start + actual_samples_per_waveform
+            mv[start:end] = waveform_data[site]
+
+            i += 1
+
+        site_list_str = ','.join(site_list)
+
+        self._write_source_waveform_site_unique_u32(site_list_str, waveform_name, len(waveform_data), actual_samples_per_waveform, data)
+