VNA.py

#!/usr/bin/python3

import pyvisa
import VNA_dummy


def create_vna(hostname, target_accuracy):
    if hostname == 'localhost':
        return VNA_dummy.get_vna_dummy(target_accuracy)
    else:
        return Vna(hostname, target_accuracy)


class Vna:
    def __init__(self, ip_address, target_accuracy):
        """
                Implements the basic operation of the VNA

        """
        # open connection here

        # store the credentials so that methods can access them
        self.__ip_address = ip_address
        self.target_accuracy = target_accuracy

        # we try to connect to the VNA just to see if there is an error
        self.rm = pyvisa.ResourceManager()

        self.vna = self.rm.open_resource("TCPIP::" + self.__ip_address + "::5025::SOCKET")
        self.vna.write_termination = '\r\n'
        self.vna.read_termination = '\n'
        self.vna.timeout = 5000

        # Preset the VNA and wait for preset completion via use of *OPC?
        self.preset_vna()

        # Clear the event status registers and empty the error queue
        self.clear_status()

    def get_tuples_of_current_traces(self):
        """
        Returns list of tuples containing trace and measured parameter pairs

        :return: list of tuples
        """
        traces = self.get_current_traces()
        result = []
        for i in range(0, len(traces)):
            if i % 2 == 0:
                result.append((traces[i], traces[i + 1]))
        return result

    def get_current_traces(self):
        """
        Returns list of string containing traces and measured parameters

        :return: list of strings
        """
        return self.vna.query("CALC1:PAR:CATALOG?").replace("'", "").split(',')

    def create_new_trace(self, name, parameter):
        """
        Creates new trace

        Parameters
        ----------
        name : str
            Name of new trace.
        parameter : int
            Measured parameter.

        Returns
        -------
        None.

        """
        self.vna.write("CALC1:PAR:SDEF '" + name + "', '" + parameter + "'")

    def set_trace_measurements(self, trace, measurement):
        """
        Set measured parameter of given trace

        Parameters
        ----------
        trace : str
            Name of new trace.
        measurement : int
            Measured parameter.

        Returns
        -------
        None.

        """
        self.vna.write("CALC1:PAR:MEAS '" + trace + "', '" + measurement + "'")

    def get_current_cw_frequency(self):
        return self.vna.query('SOUR:FREQ:CW?')

    def get_current_power(self):
        return self.vna.query('SOUR:POW?')

    def set_cw_frequency(self, freq):
        return self.vna.write(f'SOUR:FREQ:CW {freq}')

    def set_power(self, power):
        return self.vna.write(f'SOUR:POW {power}')

    def set_sweep_points(self, points):
        return self.vna.query("SENS1:SWE:POIN " + str(points) + ";*OPC?")

    def check_error_queue(self):
        """
        Check the error queue. Initially *CLS asserted in beginning of program.
        The application should run from stem to stern error free. The final error
        queue query should return '+0, No Error', else the application has potentially
        caused a correctable error!.
        """
        return self.vna.query("SYST:ERR?")

    def query_identification(self):
        return self.vna.query("*IDN?")

    def set_active_trace(self, trace):
        """
        Selects active trace

        Parameters
        ----------
        trace : str
            Name of trace.
        """
        self.vna.write("CALC1:PAR:SEL '" + trace + "'")

    def set_sweep_mode(self, mode):
        """
        Trigger assertion with hold-off for trigger complete via *OPC?

        Parameters
        ----------
        mode : str
            Sweep mode.
        """
        return self.vna.query("SENS1:SWE:TYPE " + mode + "; *OPC?")

    def get_measurement_string(self, trace, result_format):
        """
        Returns string of measured values

        Parameters
        ----------
        trace : str
            Name of trace.
        result_format : str
            VNA format of returned result.

        Returns
        -------
        None.

        """
        # Set data transfer format to ASCII
        self.vna.write("FORM:DATA ASCII")
        return self.vna.query("CALC1:DATA:TRAC? '" + trace + "', " + result_format + "")

    def get_list_of_measurement_values(self, trace, result_format):
        """
        Returns list of measured values

        Parameters
        ----------
        trace : str
            Name of trace.
        result_format : str
            VNA format of returned result.

        Returns
        -------
        None.

        """
        measurements = self.get_measurement_string(trace, result_format)
        result = []
        for x in measurements.split(',')[10:]:
            if x != '0' and x != '-999':  # no value is indicated with 0 for SDAT and -999 for FDAT
                result.append(float(x))
        return result

    def clear_status(self):
        """
        Clear the event status registers and empty the error queue
        """
        self.vna.write("*CLS")

    def preset_vna(self):
        self.vna.write("SYST:PRES; *OPC?")
        self.vna.read()

    #Hack: put in the frequency which is configured in the config file to guess if loading was
    #successful. FIXME: Do proper error checking here
    def load_config(self, config_file, frequency):
        """
        Load the config from the VNA's internal drive. It must be in the folder
        C:\\Users\\Public\\Documents\\Rohde-Schwarz\\ZNA\\RecallSets\\

        Parameters
        ----------
        config_file : str
            znxml file with VNA config and calibration. Only the file name without folder.

        """
        
        #Backslashes are escaped. Default path on the ZNA.
        directory = 'C:\\Users\\Public\\Documents\\Rohde-Schwarz\\ZNA\\RecallSets\\'
        full_path = directory+config_file
        #The string has to have single quotes, so we use double quotes to define the string.
        self.vna.write("MMEM:LOAD:STAT 1,'"+full_path+"'")
        self.vna.query('*OPC?')

        #FIXME: raise exception if loading not successful

    def do_single_sweep(self):
        """
        Start a single sweep and wait until it is done.

        Notice: Due to the hard coded timeout of 5 seconds, the sweep must not take longer!
        """
        #start measurement
        self.vna.write('INIT1')
        #wait until finished
        self.vna.query('*OPC?')
        
    def close(self):
        self.vna.close()

    def __del__(self):
        self.close()