""" Code for controlling data acquisition cards (DAQ). Different models of DAQ card should inherit from the class daq (think of this like a java interface). """ from typing import Optional import datetime from re import match import json import ctypes as ct import numpy as np import matplotlib.pyplot as plt from matplotlib.figure import Figure from matplotlib.text import Text from warnings import warn try: import PyDAQmx as daqmx except ImportError: daqmx = None warn("PyDAQmx could not be imported") class daq: def __init__(self): pass def set_digital_lines_by_address(self, values, addresses): pass def set_analog_lines_by_address(self, array, addresses): pass def set_sequence(self): pass def start_sequence(self): pass def stop_sequence(self): pass class nidaq(daq): """ Class for controlling National Instruments DAQ # todo: want better way to deal with digital line addresses. # todo: Maybe should store these for each line in self.digital_lines # todo: and separately store the block indices somehow... # todo: right now stores addresses and etc. in lists and use the index of the list as the index of the lines # todo: but this could be very confusing e.g. if have port0/0:15 and then some other line like port1/1 # todo: in that case have 0, ..., 15 representing first lines and then 16 represents port1/1. # todo: so maybe better to store dictionary with direct mappings between line names and addresses? """ def __init__(self, dev_name: str = "Dev1", digital_lines: str = "port0/line0:15", analog_lines: tuple = ("ao0", "ao1", "ao2"), analog_input_lines: tuple[str] = ("ai0",), digital_line_names: Optional[dict] = None, analog_line_names: Optional[dict] = None, presets: Optional[dict] = None, config_file: str = None, component: str = Optional[None], initialize: bool = True): """ Initialize DAQ. Note that DAQ can be instantiated before the actual DAQ is present :param dev_name: device names, typically of the form `Devk` for k an integer :param digital_lines: :param analog_lines: list of analog lines :param digital_line_names: dictionary where keys give the name of the lines and values give the line index. It is not necessary for every line to have a name :param analog_line_names: :param presets: dictionary of presets :param config_file: alternative method of provided digital_line_names, analog_line_names, and presets. If config_file is supplied, these other keyword arguments should not be supplied :param component: component in json file to load configuration from :param initialize: """ super().__init__() if config_file is not None and (digital_line_names is not None or analog_line_names is not None or presets is not None): raise ValueError("config_file and either digital_line_names, analog_line_names, or presets" " were both provided. If config_file is provided, do not also provide this other info") if config_file is not None: digital_line_names, analog_line_names, presets, _ = load_config_file(config_file, component) self.dev_name = dev_name # todo: can I read lines from device like this? daqmx.GetDevTerminals() # digital output lines self.digital_lines = f"/{dev_name}/{digital_lines}" # todo: get rid of in favor of digital_lines_addresses self.n_digital_lines = 16 # todo: want to detect not hard code self.digital_lines_addresses = [f"/{dev_name:s}/port0/line{ii:d}" for ii in range(self.n_digital_lines)] self.digital_line_names = digital_line_names self.last_known_digital_val = np.zeros(self.n_digital_lines, dtype=np.uint8) self.do_re = ".*Dev(\d+).*port(\d+).*line(\d+)" # analog output lines self.analog_lines = [f"/{dev_name}/{line}" for line in analog_lines] self.n_analog_lines = len(self.analog_lines) self.analog_line_names = analog_line_names self.last_known_analog_val = np.zeros(self.n_analog_lines, dtype=float) # self.ao_re = ".*Dev(\d+).*ao(\d+)" # analog input lines self.analog_input_line_names = [f"/{dev_name}/{line}" for line in analog_input_lines] self.n_analog_inputs = len(self.analog_input_line_names) # preset states self.presets = presets # task handles self._task_do = None self._task_ao = None self._task_di = None self._task_ai = None self._task_ct = None # any code which requires device to already be present should go inside this block self.initialized = initialize if self.initialize: pass def initialize(self, **kwargs): """ :param kwargs: :return: """ self.__init__(initialize=True, **kwargs) def reset(self): """ reset device :return: """ daqmx.DAQmxResetDevice(self.dev_name) self.set_digital_once(np.zeros(self.n_digital_lines)) def get_do_address(self, dev, port, line): """ :param dev: :param port: :param line: :return: """ # todo: remove if not useful return f"Dev{dev:d}/port{port:d}/line{line:d}" def read_do_address(self, address): """ :param address: :return: """ # todo: remove if not useful m = match(self.do_re, address) if m is None: raise ValueError("") dev = int(m.group(1)) port = int(m.group(2)) line = int(m.group(3)) return dev, port, line def get_do_block_address(self, address_list): """ :param address_list: :return: """ # todo: still playing with this...remove if not useful devs, ports, lines = zip(*[self.read_do_address(ad) for ad in address_list]) if not all([d == devs[0] for d in devs]): raise ValueError() if not all([p == ports[0] for p in ports]): raise ValueError() if not list(lines) == list(range(lines[0], lines[-1] + 1)): raise ValueError() address_block = f"Dev{devs[0]:d}/port{ports[0]:d}/line{lines[0]}:{lines[-1]}" return address_block def set_digital_once(self, array: np.ndarray): """ Set digital lines as a block :param: sizes self.n_digital_lines :return: """ array = np.array(array).astype(np.uint8) if array.ndim != 1 or array.size != self.n_digital_lines: raise ValueError(f"array must have shape {self.n_digital_lines:d} but had shape {array.shape}") self._task_do = daqmx.Task() self._task_do.CreateDOChan(self.digital_lines, "", daqmx.DAQmx_Val_ChanForAllLines) self._task_do.WriteDigitalLines(1, 1, 10.0, daqmx.DAQmx_Val_GroupByChannel, array, None, None) self._task_do.StopTask() self._task_do.ClearTask() self.last_known_digital_val[:] = array def set_digital_lines_by_address(self, array: np.ndarray, addresses: list): if array.shape != (len(addresses),): raise ValueError(f"array must have shape {len(addresses):d}, but had shape {array.shape}") try: self._task_do = daqmx.Task() self._task_do.CreateDOChan(", ".join(addresses), "", daqmx.DAQmx_Val_ChanForAllLines) self._task_do.WriteDigitalLines(1, 1, 10.0, daqmx.DAQmx_Val_GroupByChannel, np.atleast_1d(array).astype(np.uint8), None, None) # cache values for ii, ad in enumerate(addresses): ind = np.nonzero([a == ad for a in self.digital_lines_addresses])[0] if len(ind) > 0: self.last_known_digital_val[ind] = array[ii] # todo: get exception type of trying to write invalid data and only catch that specific type of exception... except Exception as e: print(e) finally: self._task_do.StopTask() self._task_do.ClearTask() def set_digital_lines_by_index(self, array: np.ndarray, lines: Optional[list] = None): """ Set digital lines :param array: 1D array of same size as number of lines you want to set :param lines: index of lines to set :return: """ if lines is None: lines = list(range(self.n_digital_lines)) if array.shape != (len(lines),): raise ValueError(f"array must have shape {len(lines):d}, but had shape {array.shape}") addresses = [self.digital_lines_addresses[l] for l in lines] return self.set_digital_lines_by_address(array, addresses) def set_digital_lines_by_name(self, array: np.ndarray, line_names: list): """ Set digital lines by name # todo: better to take dictionary as argument? {line: value} :param array: :param line_names: :return: """ if self.digital_line_names is None: raise ValueError("cannot set lines by name because self.digital_line_names is None") lines = [self.digital_line_names[n] for n in line_names] return self.set_digital_lines_by_index(array, lines) def set_analog_lines_by_address(self, array, addresses): """ :param array: :param addresses: :return: """ raise NotImplementedError("todo: this should be main function for setting analog lines once") def set_analog_once(self, array: np.ndarray, lines: Optional[list] = None, lower_lims_volts: float = -5.0, upper_lims_volts: float = 5.0): """ Set analog lines once :param array: :param lines: :param lower_lims_volts: floating point number or array with same size as array :param upper_lims_volts: :return: """ # check lines if lines is None: lines = list(range(self.n_analog_lines)) # check array array = np.array(array) if array.shape != (len(lines),): raise ValueError(f"array must have shape {len(lines):d}, but had shape {array.shape}") # get lower voltage limits for all lines lower_lims_volts = np.atleast_1d(lower_lims_volts) if lower_lims_volts.size == 1: lower_lims_volts = np.tile(lower_lims_volts, len(lines)) # get upper voltage limits for all lines upper_lims_volts = np.atleast_1d(upper_lims_volts) if upper_lims_volts.size == 1: upper_lims_volts = np.tile(upper_lims_volts, len(lines)) # program DAQ for ii, l in enumerate(lines): try: # ensure we dispose of task properly even if write fails because out of range self._task_ao = daqmx.Task() self._task_ao.CreateAOVoltageChan(self.analog_lines[l], "", lower_lims_volts[ii], upper_lims_volts[ii], daqmx.DAQmx_Val_Volts, None) self._task_ao.WriteAnalogScalarF64(True, daqmx.DAQmx_Val_WaitInfinitely, array[ii], None) self.last_known_analog_val[l] = array[ii] except daqmx.InvalidAODataWriteError as e: print(e) finally: self._task_ao.StopTask() self._task_ao.ClearTask() def set_analog_lines_by_name(self, array: np.ndarray, line_names: list, lower_lims_volts: float = -5.0, upper_lims_volts: float = 5.0): """ :param array: :param line_names: :param lower_lims_volts: :param upper_lims_volts: :return: """ if isinstance(line_names, str): line_names = [line_names] if self.analog_line_names is None: raise ValueError("cannot set lines by name because self.analog_line_names is None") lines = [self.analog_line_names[n] for n in line_names] return self.set_analog_once(array, lines, lower_lims_volts, upper_lims_volts) def set_preset(self, preset_name: str): """ Set DAQ to preset value :param preset_name: :return: """ if self.presets is None: raise ValueError("cannot set presets because self.presets is None") preset = self.presets[preset_name] # set digital lines, if there are any d = preset["digital"] if len(d) > 0: d_lines, d_arr = list(zip(*d.items())) self.set_digital_lines_by_name(np.array(d_arr), d_lines) # set analog lines, if there are any a = preset["analog"] if len(a) > 0: a_lines, a_arr = list(zip(*a.items())) self.set_analog_lines_by_name(np.array(a_arr), a_lines) def set_sine_wave(self, amps: np.ndarray, offs: np.ndarray, frq: float, lines: Optional[list] = None, nsamples: int = 100): """ Generate sine waves on selected analog channels. After setting, start with start_sequence() and stop with stop_sequence() :param amps: :param offs: :param frq: :param lines: :param nsamples: :return: """ if lines is None: lines = list(range(self.n_analog_lines)) amps = np.atleast_1d(amps) offs = np.atleast_1d(offs) ts = np.linspace(0, 1, nsamples) analog_array = amps[None, :] * np.sin(2 * np.pi * ts[:, None]) + offs[None, :] self._task_ao = daqmx.Task() self._task_ao.CreateAOVoltageChan(", ".join([d for ii, d in enumerate(self.analog_lines) if ii in lines]), "", -5.0, 5.0, daqmx.DAQmx_Val_Volts, None) self._task_ao.CfgSampClkTiming("", frq * len(ts), daqmx.DAQmx_Val_Rising, daqmx.DAQmx_Val_ContSamps, analog_array.shape[0]) samples_per_ch_ct = ct.c_int32() self._task_ao.WriteAnalogF64(analog_array.shape[0], False, 10.0, daqmx.DAQmx_Val_GroupByScanNumber, analog_array, ct.byref(samples_per_ch_ct), None) def set_sequence(self, digital_array: np.ndarray, analog_array: np.ndarray, sample_rate_hz: float, digital_clock_source: str = "OnBoardClock", analog_clock_source: str = "OnBoardClock", digital_input_source: Optional[str] = None, di_export_line: Optional[str] = None, continuous: bool = True, nrepeats: Optional[int] = None, analog_read: bool = True, pause_trigger_line: str = "/Dev1/PFI12", interval: float = 0., pause_every_n: int = 1): """ Set a sequence of digital (and optionally also analog) commands to the DAQ. This sequence can be started and stopped using start_sequence() and stop_sequence() :param digital_array: array of size n x n_channels :param analog_array: array of size m x m_channels. m and n need not be equal :param sample_rate_hz: sample rate at which digital samples will be generated :param digital_clock_source: clock source used for the digital task :param analog_clock_source: clock source used for the analog task :param digital_input_source: optional external line to perform edge detection on. The edge detection signal will be routed to di_export_line :param di_export_line: The edge detection signal read from digital_input_source is routed to this line. This line can then be used as a reference trigger for another line. Typically, this is used to trigger the analog program. In that cause analog_clock_source should be set to the same value as di_export_line :param continuous: if True, then sequence will be run repeatedly. If False, sequence will be run once. :param nrepeats: useful for specifying how much data to acquire with analog read :param analog_read: read all analog in ports during sequence :param pause_trigger_line: line which is used to pause the sequence when running with an interval of time between repeats :param interval: interval of time between repeats :param pause_every_n: perform n-repeats, then wait interval, then perform n more repeats :return: examples #################### # digital + analog sequences with same number of steps >>> daq = nidaq() >>> digital_array = np.zeros((10, 16)) # create digital array >>> digital_array[::2, 12] = 1 # alternative 0 and 1 on line 12 >>> analog_array = np.zeros((10, 4)) # create analog array >>> analog_array[::2] = 3.0 # alternate all lines between 0 and 3 volts >>> dt = 100e-6 # time step in seconds >>> daq.set_sequence(digital_array, analog_array, 1/dt) >>> daq.start_sequence() >>> time.sleep(1) >>> daq.stop_sequence() # digital + analog sequence and using one of the digital lines to advance the analog sequence >>> daq = nidaq() >>> digital_array = np.zeros((10, 16)) # create digital array >>> digital_array[::2, 12] = 1 # alternative 0 and 1 on line 12. We will use this as the clock source for the analog signal >>> analog_array = np.zeros((5, 4)) # create analog array >>> analog_array[::2] = 3.0 # alternate all lines between 0 and 3 volts >>> dt = 100e-6 # time step in seconds >>> # assume that we hook up digital line 12 to the PFI1 input port >>> # We will export this signal to the PFI2 input port and use this as the clock source for the analog lines >>> daq.set_sequence(digital_array, analog_array, 1/dt, analog_clock_source="/Dev1/PFI2", digital_input_source="/Dev1/PFI1", di_export_line="/Dev1/PFI2") >>> daq.start_sequence() >>> time.sleep(1) >>> daq.stop_sequence() """ if nrepeats is None: nrepeats = 1 start_trigger = f"/{self.dev_name}/do/StartTrigger" counter = f"/{self.dev_name}/ctr0" # ###################### # digital task # ###################### self._task_do = daqmx.Task() self._task_do.CreateDOChan(self.digital_lines, "", daqmx.DAQmx_Val_ChanForAllLines) # clock source if continuous: repeat = daqmx.DAQmx_Val_ContSamps else: repeat = daqmx.DAQmx_Val_FiniteSamps self._task_do.CfgSampClkTiming(digital_clock_source, sample_rate_hz, daqmx.DAQmx_Val_Rising, repeat, digital_array.shape[0]) # configure trigger source # self._task_do.CfgDigEdgeStartTrig(start_trigger, daqmx.DAQmx_Val_Rising) # Write the output waveform samples_per_ch_ct_digital = ct.c_int32() self._task_do.WriteDigitalLines(digital_array.shape[0], False, 10.0, daqmx.DAQmx_Val_GroupByChannel, digital_array, ct.byref(samples_per_ch_ct_digital), None) # ###################### # create pause trigger # when running a continuous sequence, this will be used to freeze sequence for an interval after each iteration # this supports sequences with high time resolution AND long time intervals between iterations # ###################### if interval == 0: # ensure no pause trigger self._task_do.ResetPauseTrigType() interval_real = None self._task_ct = None else: # set pause trigger for digital line self._task_do.SetPauseTrigType(daqmx.Val_DigLvl) self._task_do.SetDigLvlPauseTrigSrc(pause_trigger_line) self._task_do.SetDigLvlPauseTrigWhen(daqmx.Val_High) # math for interval dt = 1 / sample_rate_hz N = int(interval // dt) interval_real = N * dt n = digital_array.shape[0] # pause trigger goes high for second half of last time-step in sequence delay = (n * pause_every_n - 0.5) * dt # pause triggers stays high until the second half of the last time-step before the next cycle duty_cycle = (N - n * pause_every_n) / N if duty_cycle <= 0: raise ValueError(f"Requested interval of {N:d} time steps is shorter" f" than sequence length = {n * pause_every_n:d} steps." " You must select a longer interval.") # configure counter to be used as pause trigger self._task_ct = daqmx.Task() self._task_ct.CreateCOPulseChanFreq(counter, "", daqmx.Val_Hz, daqmx.Val_Low, delay, 1 / interval_real, duty_cycle) self._task_ct.CfgImplicitTiming(daqmx.DAQmx_Val_ContSamps, 0) self._task_ct.CfgDigEdgeStartTrig(start_trigger, daqmx.DAQmx_Val_Rising) self._task_ct.SetCOPulseTerm("", pause_trigger_line) # ###################### # set up analog trigger line # if analog trigger source is not internal, then need to set up a digital input task that perform edge detection # on the digital_input_source port. Then, the edge detections must be routed to the analog trigger source port # # currently I manually route one of the digital out lines to the digital_input_source using a wire # ###################### if digital_input_source is not None: self._task_di = daqmx.Task() self._task_di.CreateDIChan(digital_input_source, "", daqmx.DAQmx_Val_ChanForAllLines) # PFI lines are unbuffered, so must specify this self._task_di.CfgInputBuffer(0) # event on rising edge self._task_di.CfgChangeDetectionTiming(digital_input_source, None, daqmx.DAQmx_Val_ContSamps, 0) # export signal to another pin, so can use it to trigger or etc. if di_export_line is not None: self._task_di.ExportSignal(daqmx.DAQmx_Val_ChangeDetectionEvent, di_export_line) # configure start trigger self._task_di.CfgDigEdgeStartTrig(start_trigger, daqmx.DAQmx_Val_Rising) # ###################### # analog ouput task # ###################### if analog_array is None: self._task_ao = None else: # set tasks self._task_ao = daqmx.Task() self._task_ao.CreateAOVoltageChan(", ".join(self.analog_lines), "", -5.0, 5.0, daqmx.DAQmx_Val_Volts, None) self._task_ao.CfgSampClkTiming(analog_clock_source, sample_rate_hz, daqmx.DAQmx_Val_Rising, repeat, analog_array.shape[0]) self._task_ao.CfgDigEdgeStartTrig(start_trigger, daqmx.DAQmx_Val_Rising) # if analog task has only one step, then we need to add a second step, # otherwise WriteAnalogF64 will complain # This can happen if we are using a digital line to trigger the analog lines that only rarely change, if analog_array.shape[0] == 1: analog_array = np.concatenate((analog_array, analog_array), axis=0) samples_per_ch_ct = ct.c_int32() self._task_ao.WriteAnalogF64(analog_array.shape[0], False, 10.0, daqmx.DAQmx_Val_GroupByScanNumber, analog_array, ct.byref(samples_per_ch_ct), None) # ###################### # analog input task # ###################### if not analog_read: self._task_ai = None else: if self._task_ai is not None: try: self._task_ai.ClearTask() except (daqmx.DAQmxFunctions.InvalidTaskError, AttributeError): pass self._task_ai = daqmx.Task() self._task_ai.CreateAIVoltageChan(", ".join(self.analog_input_line_names), "", daqmx.DAQmx_Val_Diff, -10., 10., daqmx.DAQmx_Val_Volts, None ) self._task_ai.CfgSampClkTiming(digital_clock_source, sample_rate_hz, daqmx.DAQmx_Val_Rising, daqmx.DAQmx_Val_FiniteSamps, digital_array.shape[0] * nrepeats ) self._task_ai.CfgDigEdgeStartTrig(start_trigger, daqmx.DAQmx_Val_Rising) # todo: give option to block/wait for sequence to finish return interval_real def start_sequence(self): """ Start a sequence prepared using set_sequence() :return: """ if self._task_ct is not None: try: self._task_ct.StartTask() except daqmx.DAQmxFunctions.InvalidTaskError: pass if self._task_di is not None: try: self._task_di.StartTask() except daqmx.DAQmxFunctions.InvalidTaskError: pass if self._task_ao is not None: try: self._task_ao.StartTask() except daqmx.DAQmxFunctions.InvalidTaskError: pass if self._task_ai is not None: try: self._task_ai.StartTask() except daqmx.DAQmxFunctions.InvalidTaskError: pass if self._task_do is not None: try: self._task_do.StartTask() except daqmx.DAQmxFunctions.InvalidTaskError: pass def stop_sequence(self): """ Stop a sequence :return: """ # stop digital output task try: self._task_do.StopTask() self._task_do.ClearTask() except (daqmx.DAQmxFunctions.InvalidTaskError, AttributeError): pass # stop analog output task try: self._task_ao.StopTask() self._task_ao.ClearTask() except (daqmx.DAQmxFunctions.InvalidTaskError, AttributeError): pass # stop digital input task if self._task_di is not None: try: self._task_di.StopTask() self._task_di.ClearTask() except (daqmx.DAQmxFunctions.InvalidTaskError, AttributeError): pass # stop counter if self._task_ct is not None: try: self._task_ct.StopTask() self._task_ct.ClearTask() except (daqmx.DAQmxFunctions.InvalidTaskError, AttributeError): pass # don't stop analog task, otherwise problems # if self._task_ai is not None: # try: # self._task_ai.StopTask() # # self._task_ai.ClearTask() # except (daqmx.DAQmxFunctions.InvalidTaskError, AttributeError): # pass def read_ai(self, n_samples, timeout=1., stop=True): """ read analog input data recorded during sequence :param n_samples: :param timeout: :param stop: :return: """ n_channels = len(self.analog_input_line_names) if n_channels != 1: raise NotImplementedError() data = np.zeros((n_samples, n_channels), dtype=np.float64) arr_size = data.size read = ct.c_int32() self._task_ai.ReadAnalogF64(n_samples, # numSampsPerChan timeout, # timeout in seconds daqmx.DAQmx_Val_GroupByChannel, # fillMode data, # readArray[] arr_size, # arraySizeInSamps ct.byref(read), # sampsPerchanRead None # reserved ) if stop: self._task_ai.StopTask() return data # ########################### # helper functions for working with line mappings and program arrays # ########################### def plot_daq_program(arr: np.ndarray, line_map: Optional[dict] = None, title: str = "", **kwargs) -> Figure: """ Plot DAQ program as an array :param arr: ntimes x nchannels array :param line_map: dictionary of line names :param title: :param kwargs: :return figh: """ if line_map is None: line_names = {} k, v = zip(*list(line_map.items())) v = np.array(v) ticks = [] for ii in range(arr.shape[1]): ind = np.argwhere(v == ii) if ind.size > 1: raise ValueError() elif ind.size == 1: ticks.append(Text(float(ii), 0, k[ind[0][0]])) else: ticks.append(Text(float(ii), 0, "")) figh = plt.figure(**kwargs) ax = figh.add_subplot(1, 1, 1) ax.set_title(title) ax.imshow(arr, aspect="auto", interpolation="none") ax.set_xticks(range(arr.shape[1])) ax.set_xticklabels(ticks) ax.set_xlabel("channel") ax.set_ylabel("time step") return figh def get_line_names(map: dict) -> list: """ Given a dictionary which specifies the mapping between line names and line indices, get list of line names such that line ii is called name[ii] :param map: a dictionary describing a mapping between line names and line indices :return line_names: """ # get line names k = list(map.keys()) v = np.array(list(map.values())) ind_max = np.max(v) nchannels = ind_max + 1 line_names = [] for ii in range(nchannels): inds = np.nonzero(v == ii) if len(inds) > 1: raise ValueError() ind = inds[0] if len(inds) == 1: ind = int(ind) name = k[ind] else: name = "" line_names.append(name) return line_names def preset_to_array(preset: dict, do_map: dict, ao_map: dict, n_digital_channels: Optional[int] = None, n_analog_channels: Optional[int] = None): """ Get arrays to program daq from presets :param preset: a dictionary with two keys: "digital" and "analog". preset["digital"] is another dictionary where the keys are some subset of the keys defined in do_map, and the values are the digital and analog voltages for the preset :param do_map: digital output map dictionary, where do_map["line_name"] = line index :param ao_map: :param n_digital_channels: size used to generate array. If not specified use the largest value in do_map :param n_analog_channels: size used to generate array. If not specific use the largest value in ao_map :return digital_array, analog_array: """ # get digital array if n_digital_channels is None: n_digital_channels = max(list(do_map.values())) + 1 digital_array = np.zeros((n_digital_channels), dtype=np.uint8) for name in list(preset["digital"].keys()): digital_array[do_map[name]] = preset["digital"][name] # get analog array if n_analog_channels is None: n_analog_channels = max(list(ao_map.values())) + 1 analog_array = np.zeros((n_analog_channels)) for name in list(preset["analog"].keys()): analog_array[ao_map[name]] = preset["analog"][name] return digital_array, analog_array def format_config_data(digital_map: dict, analog_map: dict, presets: dict): """ Prepare data to be saved in configuration file. This helper function is useful if you want to save this data as one entry in a larger configuration file :param digital_map: :param analog_map: :param presets: :return: """ now = datetime.datetime.now() tstamp = f"{now.year:04d}_{now.month:02d}_{now.day:02d}_{now.hour:02d};{now.minute:02d};{now.second:02d}" data = {"timestamp": tstamp, "analog_map": analog_map, "digital_map": digital_map, "presets": presets} return data def save_config_file(fname: str, digital_map: dict, analog_map: dict, presets: dict): """ Save configuration data to json file :param fname: :param analog_map: :param digital_map: :param presets: :return: """ data = format_config_data(digital_map, analog_map, presets) with open(fname, "w") as f: json.dump(data, f, indent="\t") def load_config_file(fname: str, component: Optional[str] = None) -> (dict, dict, dict, str): """ load configuration data from json file :param fname: :param component: :return analog_map, digital_map, presets, tstamp: """ with open(fname, "r") as f: data = json.load(f) if component is not None: data = data[component] tstamp = data["timestamp"] analog_map = data["analog_map"] digital_map = data["digital_map"] presets = data["presets"] return digital_map, analog_map, presets, tstamp if __name__ == "__main__": # example configuration # DAQ line mapping daq_do_map = {"odt_cam": 0, "odt_shutter": 9, "odt_laser": 2, "sim_cam": 5, "sim_shutter": 6, "blue_laser": 8, "green_laser": 7, "red_laser": 1, "dmd_enable": 4, "dmd_advance": 3, "odt_cam_enable": 10, "signal_monitor": 11, "analog_trigger": 12 } # analog lines daq_ao_map = {"vc_mirror_x": 0, "vc_mirror_y": 1, "z_stage": 2, "z_stage_monitor": 3} # dictionary of presets # each preset is a dictionary with keys "digital" and "analog" # preset["blue"]["digital"] is a dictionary where the keys are the channel names presets = {"off": {"digital": {"odt_cam": 0, "odt_shutter": 0, "odt_laser": 0, "sim_cam": 0, "sim_shutter": 0, "blue_laser": 0, "green_laser": 0, "red_laser": 0, "dmd_enable": 0, "dmd_advance": 0}, "analog": {} }, "blue": {"digital": {"odt_shutter": 0, "odt_laser": 0, "sim_shutter": 1, "blue_laser": 1, "green_laser": 0, "red_laser": 0}, "analog": {"vc_mirror_x": -0.25, "vc_mirror_y": 0}}, "green": {"digital": {"odt_shutter": 0, "odt_laser": 0, "sim_shutter": 1, "blue_laser": 0, "green_laser": 1, "red_laser": 0}, "analog": {"vc_mirror_x": -4.75, "vc_mirror_y": -1.05}}, "red": {"digital": {"odt_cam": 0, "odt_shutter": 0, "sim_shutter": 1, "blue_laser": 0, "green_laser": 0, "red_laser": 1}, "analog": {"vc_mirror_x": -0.7, "vc_mirror_y": 0}}, "odt": {"digital": {"odt_shutter": 1, "odt_laser": 1, "sim_shutter": 0, "blue_laser": 0, "green_laser": 0, "red_laser": 0}, "analog": {}} } save_config_file("daq_config_example.json", daq_do_map, daq_ao_map, presets)