5. To Run a Test Program
The following Python script runs an existing Test Program. The script shows that the :JOBS SCPI subsystem is used to return and parse the measurement results. Notice that the visa address selects hislip1 for Station 1.
This Test Program runs without modification regardless of the measurements performed, settings selected , impairments used, or if user instruments measurements are included.
Copy
#******************************************************************************
# MIT License
# Copyright(c) 2023 Keysight Technologies
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#******************************************************************************
import pyvisa as visa # import VISA library
visa_address = 'TCPIP0::localhost::hislip1,4880::INSTR' # Address of Station 1
STATIONNUM = '1' # Station number to test
class Station(object):
""" Represents a Station child process. """
def __init__(self, visa_manager, number):
self.app = visa_manager
self.number = str(number) # Station number shown on FlexOTO GUI Tab
self.job_ids = list() # Job IDs returned by ":TPRogram:RUN?" command
def open_station_connection(address, station_number):
""" Returns a FlexOTO station visa object. """
rm = visa.ResourceManager()
connection = rm.open_resource(address)
connection.timeout = 20000 # Set connection timeout to 20s
connection.read_termination = '\n'
connection.write_termination = '\n'
station = Station(connection, station_number)
print('VISA connection to Session ' + station_number + ' established.', flush=True)
return station
def run_station(station):
""" Runs Session and saves a list of Job IDs in station object. Job IDs are used later to identify
measurement results. """
job_ids = station.app.query(':TPRogram:RUN?').split(',')
print('Session ' + station.number + ' Test Program started. Acquiring and analyzing data. Please wait.', flush=True)
return job_ids
def monitor_station(station):
""" Returns after all measurement acquisitions and analysis is complete. """
print('Waiting for Session ' + station.number + ' standard measurements to complete...')
station.app.query(':JOBS:ACQuire:COMPlete?')
station.app.query('*OPC?') # Wait for all jobs to be done with analysis.
return
def get_jobid_measurement_units(station, job):
""" """
tokens = {'WATT': 'W', 'DBM': 'dBm', 'DEC': 'dB', 'PERC': '%', 'RAT': ': 1'}
punits = tokens[station.app.query(':JOBS:CONFig:PUNits? ' + job)]
erunits = tokens[station.app.query(':JOBS:CONFig:ERUNits? ' + job)]
return punits, erunits
def add_measurement_units_to_result(meas, punits, erunits):
""" Measurement units are returned for all related measurement in the same Job ID. If you want different units
in two related commends (for example, average and peak-peak powers), put the measurement on different
Test Program lines which results in different Job IDs. """
measurement = meas['Value']
if 'TDECQ' in meas['Name']:
measurement += ' dB'
elif 'Ceq' in meas['Name']:
measurement += ' dB'
elif 'Outer OMA' in meas['Name']:
measurement += ' ' + punits
elif 'Outer ER' in meas['Name']:
measurement += ' ' + erunits
elif 'RLM (802.3 A_120D)' in meas['Name']: # Linearity
measurement += ''
elif 'RLM (802.3 CL_94)' in meas['Name']: # Linearity
measurement += ''
elif 'RLM (CEI)' in meas['Name']: # Linearity
measurement += ''
elif 'Level 0' in meas['Name']:
measurement += ' W'
elif 'Level 1' in meas['Name']:
measurement += ' W'
elif 'Level 2' in meas['Name']:
measurement += ' W'
elif 'Level 3' in meas['Name']:
measurement += ' W'
elif 'Trans. Time' in meas['Name']:
measurement += ' ps'
elif 'Overshoot' in meas['Name']:
measurement += ' %'
elif 'Undershoot' in meas['Name']:
measurement += ' %'
elif 'Power Excursion' in meas['Name']:
measurement += ' ' + punits
elif 'Average Power' in meas['Name']:
measurement += ' ' + punits
elif 'Pk-Pk Power' in meas['Name']:
measurement += ' ' + punits
return measurement
def get_results_for_Job_ID(station, job):
""" For Job ID, returns a measurement results string and converts the string to a list of
measurement results dictionaries. Fixture/lane dict followed by one dict for each measurement.
Convert:
'Fixture=DUT Fixture 1,Lane=Lane 4;Name=TDECQ,Value=1E-2,Status=Correct;'
To:
[{'Fixture': 'DUT Fixture 1','Lane': 'Lane 4'},
{'Name': 'TDECQ','Value': '1E-2, 'Status':, 'Correct'},
{ ...}]
"""
meas_list = station.app.query(':JOBS:RESults? ' + job).split(';')
for i in range(0, len(meas_list)):
measurement_fields = meas_list[i].split(',') # Results list of one measurement
for n in range(0, len(measurement_fields)):
measurement_fields[n] = measurement_fields[n].split('=') # eg, ['Name','TDECQ']
meas_list[i] = dict(measurement_fields) # create dictionary with Name, Value, and Status keys
measurements = list()
for meas in meas_list: # Remove non-measurements from results
values = meas.values()
if ('Waveform' in values) or ('Eye Diagram' in values):
continue
measurements.append(meas)
return measurements
def print_results_for_Job_ID(station, job, measurements):
""" Prints results for Job ID. """
user_meas_names = {'User Meas1', 'User Meas2'}
punits, erunits = get_jobid_measurement_units(station, job)
fixture_lane = measurements.pop(0)
fixture = fixture_lane['Fixture']
lane = fixture_lane['Lane']
print('\n\tJob: ' + job + ',\tFixture: "' + fixture + '"\t"' + lane + '"')
print()
for meas in measurements:
meas_with_units = add_measurement_units_to_result(meas, punits, erunits)
if 'Correct' in meas['Status']:
print('\t\t' + fixture + ', ' + lane + ': ' + meas['Name'] + ': ' + meas_with_units)
else:
print('\t\t' + fixture + ', ' + lane + ': "' + meas['Status'] + '"')
if meas['Name'] not in user_meas_names:
aengine = station.app.query(':JOBS:RESults:INFO:AENGine? ' + job)
amodule = station.app.query(':JOBS:RESults:INFO:AMODule? ' + job)
print('\t\t\tAcquisition Engine: ' + aengine + ' with ' + 'Acquisition Module: ' + amodule)
def save_all_results_to_file(station):
""" Saves all results in zip files. """
station.app.write(':DISK:RESults:SAVE:SELection ALL')
filename = 'Results_Session' + station.number
station.app.write(':DISK:RESults:FNAMe "' + filename + '"')
station.app.write(':DISK:RESults:SAVE')
print('\n\tSession ' + station.number + ' zip results file: ' + filename + '.zip')
def rerun_test_plan():
""" Prompt user for another Test Plan run. """
answer = input('\nRun the Test Plane again (y/n)?: ').lower()
if 'y' in answer:
return True
else:
return False
def remove_test_results(station):
""" Removes test results. """
station.app.write(':JOBS:RESults:REMove:ALL')
def close_station(station):
""" Places Session1 and Primary in local mode. """
station.app.write(':SYSTem:GTLocal')
station.app.close()
# main loop
tst_station = open_station_connection(visa_address, STATIONNUM)
while True:
tst_station.job_ids = run_station(tst_station) # returns new Job IDs
monitor_station(tst_station)
print('\nStation ' + tst_station.number + ' measurements:')
for job in tst_station.job_ids:
meas_list = get_results_for_Job_ID(tst_station, job)
print_results_for_Job_ID(tst_station, job, meas_list)
save_all_results_to_file(tst_station)
if not rerun_test_plan():
break
remove_test_results(tst_station)
close_station(tst_station)