Initial commit with READ working

Author: dan-zilla

.vscode/c_cpp_properties.json

@@ -0,0 +1,19 @@
+{
+    "configurations": [
+        {
+            "name": "Win32",
+            "includePath": [
+                "${workspaceFolder}/**"
+            ],
+            "defines": [
+                "_DEBUG",
+                "UNICODE",
+                "_UNICODE"
+            ],
+            "cStandard": "c17",
+            "cppStandard": "c++17",
+            "intelliSenseMode": "windows-msvc-x64"
+        }
+    ],
+    "version": 4
+}

.vscode/settings.json

@@ -0,0 +1,8 @@
+{
+    "python.pythonPath": "C:\\Users\\Murmann-Group\\anaconda3\\envs\\NI-RRAM-Python\\python.exe",
+    "python.linting.pylintPath": "C:\\Users\\Murmann-Group\\anaconda3\\envs\\NI-RRAM-Python\\Scripts\\pylint.exe",
+    "python.linting.enabled": true,
+    "python.linting.lintOnSave" : true,
+    "python.linting.pylintUseMinimalCheckers": false,
+    "C_Cpp.errorSquiggles": "Disabled",
+}

ni_hsdio.py

@@ -0,0 +1,120 @@
+"""Provides Python wrapper for NI-HSDIO C library DLL. Currently implements static generation."""
+from ctypes import CDLL, c_int, c_uint, create_string_buffer, pointer
+
+
+class NIHSDIOException(Exception):
+    """Exception produced by the NIHSDIO class"""
+    def __init__(self, msg):
+        super().__init__(f"NIHSDIO: {msg}")
+
+
+class NIHSDIO:
+    """A wrapper class for NI-HSDIO C DLL. Also includes functions for static assignment."""
+
+    # Logic family constants
+    NIHSDIO_VAL_5_0V_LOGIC = 5
+    NIHSDIO_VAL_3_3V_LOGIC = 6
+    NIHSDIO_VAL_2_5V_LOGIC = 7
+    NIHSDIO_VAL_1_8V_LOGIC = 8
+    NIHSDIO_VAL_1_5V_LOGIC = 80
+    NIHSDIO_VAL_1_2V_LOGIC = 81
+
+    def __init__(self, deviceID, chanMap=None, channelList="", logicFamily=NIHSDIO_VAL_1_8V_LOGIC):
+        # Store parameters
+        self.device_id = deviceID
+        self.chans = channelList
+        self.chan_map = chanMap if chanMap is not None else {}
+
+        # Create driver and VI session
+        self.driver = CDLL("lib/niHSDIO_64.dll")
+        self.sess = c_uint(0)
+
+        # Initialize generation session
+        self.init_generation_session(deviceID)
+
+        # Configure voltage level
+        self.configure_data_voltage_logic_family(channelList, logicFamily)
+
+        # Assign static channels
+        self.assign_static_channels(channelList)
+
+    def init_generation_session(self, device_id):
+        """Wrapper function: initialize HSDIO session for digital pattern generation"""
+        # Create session
+        device_id = bytes(device_id, 'utf-8')
+        err = self.driver.niHSDIO_InitGenerationSession(device_id, 0, 0, None, pointer(self.sess))
+        self.check_err(err)
+
+    def configure_data_voltage_logic_family(self, chans="", logic_family=NIHSDIO_VAL_1_8V_LOGIC):
+        """Wrapper function: configure data voltage logic family"""
+        err = self.driver.niHSDIO_ConfigureDataVoltageLogicFamily(self.sess, chans, logic_family)
+        self.check_err(err)
+
+    def assign_static_channels(self, chans=""):
+        """Wrapper function: configure channels for static digital generation"""
+        # Create session
+        err = self.driver.niHSDIO_AssignStaticChannels(self.sess, chans)
+        self.check_err(err)
+
+    def write_static(self, write_data, channel_mask=0xFFFFFFFF):
+        """Wrapper function: write static data to configured channels"""
+        err = self.driver.niHSDIO_WriteStaticU32(self.sess, write_data, channel_mask)
+        self.check_err(err)
+
+    def check_err(self, err):
+        """Wrapper function: given an error code, return error description"""
+        if err != 0:
+            err_desc = create_string_buffer(1024)
+            self.driver.niHSDIO_GetError(None, pointer(c_int(err)), 1024, err_desc)
+            raise NIHSDIOException(err_desc.value)
+
+    def close(self):
+        """Wrapper function: close NI-HSDIO session"""
+        self.check_err(self.driver.niHSDIO_close(self.sess))
+
+    def write_data_across_chans(self, chans, data, debug=False):
+        """Writes a pattern across channels. If chans is a string, it is interpreted as a
+        channel map key. If chans is a list, it will be directly interpreted as the channels.
+        data will be interpreted as an unsigned 32-bit integer."""
+        # Type checking and conversion
+        if isinstance(chans, str):
+            chans = self.chan_map[chans]
+        if not isinstance(chans, list):
+            err = f"Channels must be specified as (or mapped to) a list: got {repr(chans)}."
+            raise NIHSDIOException(err)
+
+        # Convert data to binary string for bit access
+        data = format(data, '032b')[::-1]
+
+        # Reorder bits for static generation
+        write_data = 0
+        mask = 0
+        for chan, bit in zip(chans, data):
+            mask |= (1 << chan)
+            write_data |= (int(bit) << chan)
+
+        # Debug statements
+        if debug:
+            print("Write data:", write_data)
+            print("Mask:", mask)
+            print("Original data binary:", data[::-1])
+            print("Write data binary:", format(write_data, '032b'))
+            print("Mask binary:", format(mask, '032b'))
+            print()
+
+        # Write data using driver
+        self.write_static(write_data, mask)
+
+    def __del__(self):
+        # Try to close session on deletion
+        try:
+            self.close()
+        except NIHSDIOException:
+            pass
+
+if __name__=="__main__":
+    # Initialize and run basic static generation tests
+    nihsdio = NIHSDIO("Dev7", {"wl_addr": [1,3,5,7,9,11]})
+    nihsdio.write_static(0xFFFFFFFF)
+    nihsdio.write_data_across_chans("wl_addr", 10, debug=True)
+    nihsdio.write_data_across_chans([13,15,30,28,26,24,22], 10, debug=True)

ni_rram.py

@@ -0,0 +1,220 @@
+"""Defines the NI RRAM controller class"""
+import json
+import time
+import warnings
+import nidaqmx
+import nidcpower
+import nifgen
+import numpy as np
+from ni_hsdio import NIHSDIO
+
+warnings.filterwarnings("error")
+
+def accurate_delay(delay):
+    """Function to provide accurate time delay"""
+    _ = time.perf_counter() + delay
+    while time.perf_counter() < _:
+        pass
+
+
+class NIRRAMException(Exception):
+    """Exception produced by the NIRRAM class"""
+    def __init__(self, msg):
+        super().__init__(f"NIRRAM: {msg}")
+
+
+class NIRRAM:
+    """The NI RRAM controller class that controls the instrument drivers."""
+    def __init__(self, chip, settings="settings.json"):
+        # If settings is a string, load as JSON file
+        if isinstance(settings, str):
+            with open(settings) as settings_file:
+                settings = json.load(settings_file)
+
+        # Ensure settings is a dict
+        if not isinstance(settings, dict):
+            raise NIRRAMException(f"Settings should be a dict, got {repr(settings)}.")
+
+        # TODO: initialize logger
+
+        # Store/initialize parameters
+        self.chip = chip
+        self.settings = settings
+        self.addr = 0
+        self.prof = {"READs": 0, "SETs": 0, "RESETs": 0}
+
+        # Initialize NI-HSDIO driver
+        self.hsdio = NIHSDIO(**settings["HSDIO"])
+
+        # Initialize NI-DAQmx driver for READ voltage
+        self.read_chan = nidaqmx.Task()
+        self.read_chan.ai_channels.add_ai_voltage_chan(settings["DAQmx"]["chanMap"]["read_ai"])
+        self.read_chan.timing.cfg_samp_clk_timing(settings["READ"]["read_clk_rate"])
+
+        # Initialize NI-DAQmx driver for WL voltages
+        self.wl_ext_chans = []
+        for chan in settings["DAQmx"]["chanMap"]["wl_ext"]:
+            task = nidaqmx.Task()
+            task.ao_channels.add_ao_voltage_chan(chan)
+            task.timing.cfg_samp_clk_timing(settings["samp_clk_rate"])
+            task.start()
+            self.wl_ext_chans.append(task)
+
+        # Initialize NI-DCPower driver for BL voltages
+        self.bl_ext_chans = []
+        for chan in settings["DCPower"]["chans"]:
+            sess = nidcpower.Session(settings["DCPower"]["deviceID"], chan)
+            sess.voltage_level = 0
+            sess.commit()
+            sess.initiate()
+            self.bl_ext_chans.append(sess)
+
+        # Initialize NI-FGen driver for SL voltage
+        self.sl_ext_chan = nifgen.Session(settings["FGen"]["deviceID"])
+        self.sl_ext_chan.output_mode = nifgen.OutputMode.FUNC
+        self.sl_ext_chan.configure_standard_waveform(nifgen.Waveform.DC, 0.0, frequency=10000000)
+        self.sl_ext_chan.initiate()
+
+        # Set address to 0
+        self.set_address(0)
+
+    def read(self):
+        """Perform a READ operation. Returns tuple with (res, cond, meas_i, meas_v)"""
+        # Increment the number of READs
+        self.prof["READs"] += 1
+
+        # Address decoder enable
+        self.decoder_enable()
+
+        # Set voltages
+        self.set_vsl(0)
+        self.set_vbl(self.settings["READ"]["VBL"])
+        self.set_vwl(self.settings["READ"]["VWL"])
+
+        # Settling time for VBL
+        accurate_delay(self.settings["READ"]["settling_time"])
+
+        # Measure
+        meas_v = np.mean(self.read_chan.read(self.settings["READ"]["n_samples"]))
+        meas_i = meas_v/self.settings["READ"]["shunt_res_value"]
+        res = np.abs(self.settings["READ"]["VBL"]/meas_i - self.settings["READ"]["shunt_res_value"])
+        cond = 1/res
+
+        # Turn off VBL and VWL
+        self.set_vbl(0)
+        self.set_vwl(0)
+
+        # Address decoder disable
+        self.decoder_disable()
+
+        # TODO: log READ operation
+
+        # Return measurement tuple
+        return res, cond, meas_i, meas_v
+
+    def set_pulse(self, vwl=None, vbl=None, pw=None):
+        
+
+    def set_vsl(self, voltage):
+        """Set VSL using NI-FGen driver"""
+        # Set DC offset to V/2 since it is doubled by FGen for some reason
+        self.sl_ext_chan.func_dc_offset = voltage/2
+
+    def set_vbl(self, voltage):
+        """Set (active) VBL using NI-DCPower driver (inactive disabled)"""
+        # LSB indicates active BL channel
+        active_bl_chan = (self.addr >> 0) & 0b1
+        active_bl = self.bl_ext_chans[active_bl_chan]
+        inactive_bl = self.bl_ext_chans[1-active_bl_chan]
+
+        # Set voltages and commit
+        active_bl.voltage_level = voltage
+        active_bl.commit()
+        inactive_bl.voltage_level = 0
+        inactive_bl.commit()
+
+    def set_vwl(self, voltage):
+        """Set (active) VWL using NI-DAQmx driver (inactive disabled)"""
+        # Select 8th and 9th bit to get the channel and the driver card, respectively
+        active_wl_chan = (self.addr >> 8) & 0b1
+        active_wl_dev = (self.addr >> 9) & 0b1
+        active_wl = self.wl_ext_chans[active_wl_dev]
+        inactive_wl = self.wl_ext_chans[1-active_wl_dev]
+
+        # Write
+        active_wl.write([(1-active_wl_chan)*voltage, active_wl_chan*voltage], auto_start=True)
+        inactive_wl.write([0,0], auto_start=True)
+
+    def pulse_vwl(self, voltage, pw):
+        """Pulse (active) VWL using NI-DAQmx driver (inactive disabled)"""
+        # Select 8th and 9th bit to get the channel and the driver card, respectively
+        active_wl_chan = (self.addr >> 8) & 0b1
+        active_wl_dev = (self.addr >> 9) & 0b1
+        active_wl = self.wl_ext_chans[active_wl_dev]
+        inactive_wl = self.wl_ext_chans[1-active_wl_dev]
+
+        # Write
+        active_wl.write([(1-active_wl_chan)*voltage, active_wl_chan*voltage], auto_start=True)
+        inactive_wl.write([0,0], auto_start=True)
+
+    def decoder_enable(self):
+        """Enable decoding circuitry using digital signals"""
+        self.hsdio.write_data_across_chans("wl_dec_en", 0b11)
+        self.hsdio.write_data_across_chans("sl_dec_en", 0b1)
+        self.hsdio.write_data_across_chans("wl_clk", 0b1)
+
+    def decoder_disable(self):
+        """Disable decoding circuitry using digital signals"""
+        self.hsdio.write_data_across_chans("wl_dec_en", 0b00)
+        self.hsdio.write_data_across_chans("sl_dec_en", 0b0)
+        self.hsdio.write_data_across_chans("wl_clk", 0b0)
+
+    def set_address(self, addr):
+        """Set the address and hold briefly"""
+        # Update address
+        self.addr = addr
+
+        # Extract SL and WL addresses
+        sl_addr = (self.addr >> 1) & 0b1111111
+        wl_addr = (self.addr >> 10) & 0b111111
+
+        # Write addresses to corresponding HSDIO channels
+        self.hsdio.write_data_across_chans("sl_addr", sl_addr)
+        self.hsdio.write_data_across_chans("wl_addr", wl_addr)
+        accurate_delay(self.settings["addr_hold_time"])
+
+        # Reset profiling counters
+        self.prof = {"READs": 0, "SETs": 0, "RESETs": 0}
+
+    def close(self):
+        """Close all NI sessions"""
+        # Close NI-HSDIO
+        self.hsdio.close()
+
+        # Close NI-DAQmx AI
+        self.read_chan.close()
+
+        # Close NI-DAQmx AOs
+        for task in self.wl_ext_chans:
+            task.stop()
+            task.close()
+
+        # Close NI-DCPower
+        for sess in self.bl_ext_chans:
+            sess.abort()
+            sess.close()
+
+        # Close NI-FGen
+        self.sl_ext_chan.abort()
+        self.sl_ext_chan.close()
+
+        # Give some time to shutdown
+        time.sleep(1)
+
+if __name__=="__main__":
+    # Basic test
+    nirram = NIRRAM("Chip9")
+    for address in range(2000):
+        nirram.set_address(address)
+        if address % 100 == 0:
+            print((address, nirram.read()),)