Example 6. Adding Functions
Meas. mode:
Scope
Eye
TDR
Package License:
L-RND
L-SNT
This example is the same script described in Example 5. A User Measurement. The only difference the addition of functions to encourage reuse of code snippets. Instead of using global variable which can cause problems, all variables and data structures are passed as function arguments.
Required Files
PreEmphasis-with-functions-setup.py
This script configures FlexDCA to required state for the example script. The script installs a simulated module, selects the proper FlexDCA mode, and turns on two dependent measurements: Amplitude and Peak-Peak Amplitude.
Copy
PreEmphasis-with-functions-setup.py
import pyvisa as visa # import VISA library
visa_address = 'TCPIP0::localhost::hislip0,4880::INSTR'
CHANNEL = '1A'
TIMEOUT = 10000
def open_flexdca_connection(address):
""" Opens visa connection to FlexFlexDCA. """
print('Connecting to Flexdca ...')
try:
rm = visa.ResourceManager()
connection = rm.open_resource(address)
connection.timeout = TIMEOUT # Set connection timeout to 20s
connection.read_termination = '\n'
connection.write_termination = '\n'
inst_id = connection.query('*IDN?')
print('\nFlexDCA connection established to:\n' + inst_id, flush=True)
connection.write(':SYSTem:DEFault')
connection.query('*OPC?')
except (visa.VisaIOError, visa.InvalidSession):
print('\nVISA ERROR: Cannot open instrument address.\n', flush=True)
return None
except Exception as other:
print('\nVISA ERROR: Cannot connect to instrument:', other, flush=True)
print('\n')
return None
return connection
def all_channels_off(FlexDCA):
""" Turns all available channels off. """
for slot in '12345678':
for letter in 'ABCD':
channel = slot + letter
FlexDCA.write(':CHANnel' + channel + ':DISPlay OFF')
def install_simulated_module(FlexDCA, channel, model, signal='NRZ'):
""" Simplified installation of a simulated FlexDCA module.
model
Type of simulated module. "DEM" is a dual-electrical module.
"QEM" is a quad-electrical module. "DOM" is a dual-optical
module. "QEM" is a electrical-optical module.
signal
Format of signal. NRZ or PAM4.
"""
slot = channel[0]
FlexDCA.write(':EMODules:SLOT' + slot + ':SELection ' + model)
if signal in 'NRZ':
FlexDCA.write(':SOURce' + channel + ':FORMat NRZ')
else:
FlexDCA.write(':SOURce' + channel + ':FORMat PAM4')
FlexDCA.write(':SOURce' + channel + ':DRATe 9.95328E+9')
FlexDCA.write(':SOURce' + channel + ':WTYPe DATA')
FlexDCA.write(':SOURce' + channel + ':PLENgth 127')
FlexDCA.write(':SOURce' + channel + ':AMPLitude 90E-3')
FlexDCA.write(':SOURce' + channel + ':NOISe:RN 3.0E-6')
FlexDCA.write(':SOURce' + channel + ':JITTer:RJ 4.0E-12')
FlexDCA.write(':CHANnel' + channel + ':DISPlay ON')
def return_scalar(FlexDCA, query):
""" Returns a FlexDCA scalar measurement result. """
import time
class ScalarMeasQuestionableException(Exception):
""" A scalar measurement result is questionable. """
pass
class ScalarMeasTimeOutException(Exception):
""" A scalar measurement has timed out. """
pass
timeout = FlexDCA.timeout / 1000 # convert ms to s
query = query.strip('?')
start_time = time.time() # time in seconds
while(True):
time.sleep(0.2)
status = FlexDCA.query(query + ':STATus?')
if 'CORR' in status: # valid measurement result
return FlexDCA.query(query + '?') # get measurement
elif 'QUES' in status: # questionable results
s = query + '\nReason: ' + FlexDCA.query(query + ':STATus:REASon?')
raise ScalarMeasQuestionableException(s)
elif (int(time.time() - start_time)) > timeout: # IO timout
s = query + '\nReason: ' + FlexDCA.query(query + ':STATus:REASon?')
raise ScalarMeasTimeOutException(s)
flexdca = open_flexdca_connection(visa_address)
flexdca.query(':SYSTem:DEFault;*OPC?')
all_channels_off(flexdca)
install_simulated_module(flexdca, CHANNEL, 'DEM')
flexdca.write(':ACQuire:RUN')
flexdca.query(':SYSTem:MODE OSCilloscope;*OPC?') # Switch to Eye/Mask mode
flexdca.query(':SYSTEM:AUToscale;*OPC?')
flexdca.query(':TRIGger:PLOCk ON;*OPC?')
flexdca.write(':ACQuire:RLENgth:MODE MANual')
flexdca.write(':ACQuire:RLENgth 1024')
flexdca.write(':ACQuire:SMOothing AVERage')
flexdca.write(':TIMebase:SCALe 5.000E-10')
flexdca.write(':MEASure:OSCilloscope:VAMPlitude')
flexdca.write(':MEASure:OSCilloscope:VPP')
flexdca.write(':SYSTem:GTLocal')
flexdca.close()
PreEmphasis-with-functions.xml
Installs user measurement.
Copy
PreEmphasis-with-functions.xml
<?xml version="1.0" encoding="utf-8"?>
<Measurement xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
Application="FlexDCA" MinimumSoftwareVersion="5.00.328">
<Name>Pre-Emphasis</Name>
<Abbreviation>PreEmp</Abbreviation>
<Comments>In Scope mode, measures pre-emphasis on a signal.</Comments>
<Script>PreEmphasis-with-functions.py</Script>
<Icon></Icon>
<MeasurementType>1 Source</MeasurementType>
<ValidMode>Scope</ValidMode>
<Dependent Name = "Amplitude">
<SourceIndex>1</SourceIndex>
</Dependent>
<Dependent Name = "Peak-Peak">
<SourceIndex>1</SourceIndex>
</Dependent>
<Variable Name = "MinSampl" Value = "1000" />
<Variable Name = "MaxSampl" Value = "2000" />
</Measurement>
PreEmphasis-with-functions.py
User measurement script.
Copy
PreEmphasis-with-functions.py
import math
def algorithm(inputVars):
"""
Receives FlexDCA variable structure, calculates a new user measurement, and
returns the measurement value and status.
"""
Result = 0.0
Units = 'dB'
# Status is set to 'Correct' after Result is computed.
statusDict = {'Status': 'Invalid', 'ErrorMsg': ''}
mode = 'OSCilloscope' # EYE, OSCilloscope, or TDR
statusDict = confirm_measurement_dicts(inputVars, statusDict, mode)
if statusDict['ErrorMsg']:
return statusDict
num_samples, statusDict = confirm_waveform_data(inputVars, statusDict)
if statusDict['ErrorMsg']:
return statusDict
Vpp, Vamptd, statusDict = get_dependent_measurements(inputVars,
statusDict,
mode)
if statusDict['ErrorMsg']:
return statusDict
meas_list = [Vpp, Vamptd]
statusDict = check_dependent_measurements_status(meas_list, statusDict)
if 'Invalid' in statusDict['Status']:
return statusDict
statusDict = check_number_of_samples(num_samples, inputVars, statusDict)
# Compute result
Result = 20 * math.log10(Vpp['Result'] / Vamptd['Result'])
# Return dictionary of results to FlexDCA
return { 'Result' : Result, # Calculated scalar result
'Units' : Units, # Measurement units
'Status' : statusDict['Status'], # 'Correct', 'Questionable', 'Invalid'
'ErrorMsg' : statusDict['ErrorMsg'] # Error message string
}
def confirm_measurement_dicts(inputVars, statusDict, mode):
"""
Check to see if inputVars dictionary's 'MeasData' key exists. If missing,
adds an error message to the status dictionary's message entry.
By default, the status dictionary's status is set to 'Invalid'.
FlexDCA's measurement mode is included in the error message.
"""
if 'MeasData' in inputVars:
return statusDict
else:
statusDict['ErrorMsg'] = 'Unable to retrieve any measurement data.\n' \
'Please use ' + mode + ' Mode and load the XML configuration ' \
'file rather than loading the Python measurement script directly.'
return statusDict
def confirm_waveform_data(inputVars, statusDict):
""" Check to see if waveform data (inputVars dictionary's 'SrcData' key)
contains a waveform data. If missing, adds an error message to the status
dictionary's message entry. By default, the status dictionary's status is
set to 'Invalid'. Returns number of samples found in data and statusDict.
The returned num_samples can be ignored by calling function or used to
check against desired sample limits.
"""
SrcData = inputVars['SrcData']
num_samples = len(SrcData)
if num_samples == 0:
statusDict['ErrorMsg'] = 'No waveform data.'
return (num_samples, statusDict)
def get_dependent_measurements(inputVars, statusDict, mode):
"""
Iterate through MeasData (list of dictionaries) to get dictionary for
each dependent measurement. Returns statusDict and dictionary for each
dependent measurement.
"""
MeasData = inputVars['MeasData']
Vpp, Vamptd = None, None
for meas in MeasData:
if (meas['Source1'] == inputVars['Source']):
if (meas['Name'] == 'Peak-Peak'):
Vpp = meas # Vpp is dictionary of measurement info
elif (meas['Name'] == 'Amplitude'):
Vamptd = meas # Vamptd is dictionary of measurement info
if (Vpp is None) or (Vamptd is None):
statusDict['ErrorMsg'] = 'Unable to retrieve "Peak-Peak" and ' \
'"Amplitude" measurements.\n Use ' + mode + ' Mode and ' \
'include these measurements as Dependents in your XML ' \
'configuration file.'
return (Vpp, Vamptd, statusDict)
def check_dependent_measurements_status(meas_list, statusDict):
"""
Query the status of each dependent measurement used in algorithm. If all
correct, perform user measurement. If any invalid, report an error.
If one or more are not correct, but none are invalid, report a
questionable status which does not end measurement.
"""
correct = 0
invalid = 0
questionable = 0
for measurement in meas_list:
if measurement['Status'] == 'Correct':
correct += 1
elif measurement['Status'] == 'Invalid':
invalid += 1
else:
questionable += 1
if correct == len(meas_list):
statusDict['Status'] = 'Correct'
elif invalid:
statusDict['ErrorMsg'] = 'Dependent measurements have invalid statuses.'
return statusDict
else:
statusDict['Status'] = 'Questionable'
statusDict['ErrorMsg'] = 'Dependent measurements have questionable statuses.'
return statusDict
def check_number_of_samples(num_samples, inputVars, statusDict):
"""
MinSampl and MaxSampl set the limits on how many samples per acquisition are
ideal. If the number are outside these limits, the measurement is marked
questionable. This does not halt the measurement. Demonstrates unique
variables defined in XML configuration file. Requires MinSampl and MaxSampl
be declared in xml configuration file. Requires num_samples returned by
confirm_waveform_data().
"""
MinSampl = int(inputVars['MinSampl'])
MaxSampl = int(inputVars['MaxSampl'])
if (num_samples < MinSampl) or (num_samples > MaxSampl):
statusDict['Status'] = 'Questionable'
statusDict['ErrorMsg'] = 'The number of samples per acquisition is ' \
'not within the specified range of {0} to {1}.\n Please edit ' \
'measurement script or change the number of samples per ' \
'acquisition.'.format(MinSampl, MaxSampl)
return statusDict