Test Station Setup
The test-station.py Python script demonstrates configuring and running a Station's FlexOTO Test Program for a single Test Station. Before you can run this script, you will need the have run either the hardware-bootup-N7734A.py or hardware-bootup.py scripts to configure the Hardware Diagram. However, any hardware diagram that will results in Station 1 measuring 4 lanes will work.
The following variables that are located at the top of the script to control the setup:
SESSIONNUM. This is the Station number which is associated with the hislip value in thevisa_addressstring.visa_address. If you're running the script on Test Station's PC, uselocalhost. Otherwise, use the name of the PC where FlexOTO is installed. The string includes the hislip value in theSESSIONNUMstring variable.
This script performs the following tasks:
- Configures Station 1.
- Runs Station 1's Test Program and monitors the Station's run state.
- Displays measurements and saves a zip file of the results.
The script uses a Station class to encapsulate Station's VISA connection, Station number (a string), and list of Job IDs. The Job IDs are assigned by list that is returned by the run_station() function.
Example Script
Copy
test-station.py
#******************************************************************************
# 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
STATIONNUM = '1' # Station number to test
# visa_address for Session
visa_address = 'TCPIP0::localhost::hislip' + STATIONNUM + ',4880::INSTR' # Edit as needed
class Station(object):
""" Represents a Session child process. """
def __init__(self, visa_manager, number):
self.app = visa_manager
self.number = str(number) # Session 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 Session object. Initializes Session with a visa object for the Session, the
station number (integer), and a list of Session measurements. Sets Session to its default setup. """
print('Connecting to FlexOTO Session...')
try:
rm = visa.ResourceManager()
connection = rm.open_resource(address)
connection.timeout = 20000 # Set connection timeout to 20s
connection.read_termination = '\n'
connection.write_termination = '\n'
inst_id = connection.query('*IDN?')
print('\nFlexOTO connection established to:\n' + inst_id, flush=True)
except (visa.VisaIOError, visa.InvalidSession):
print('\nVISA ERROR: Cannot open instrument address.\n', flush=True)
except Exception as other:
print('\nVISA ERROR: Cannot connect to instrument:', other, flush=True)
print('\n')
station = Station(connection, station_number)
print('VISA connection to Session ' + station_number + ' established.', flush=True)
return station
def set_fixture_lane(station):
""" Selects all DUT lanes for measurements. """
station.app.write(':TPRogram:SETup:FIXTure' + station.number + ' ENABled')
def set_waveform(station):
""" If the opticl switch supports a wavelength setting. """
station.app.write(':TPRogram:SETup:SRATe 5.3125000E+10')
station.app.write(':TPRogram:SETup:PLENgth 65535')
def set_presets(station):
""" Select FlexDCA setting presets. FlexOTO configures FlexDCA. """
station.app.write(':TPRogram:SETup:CRPReset "IEEE 802.3bs/cd/ck (53 GBd)"')
station.app.write(':TPRogram:SETup:TMPReset "IEEE 802.3cd"')
station.app.write(':TPRogram:SETup:TEPReset "IEEE 802.3cd"')
def set_standard_measurements(station):
""" Selects measurements to perform on all selected lanes. Includes saving waveform data and screen image. """
measurements = 'TDEQ,CEQ,OOMA,OER,LINearity,LEVels,TTIMe,OVERshoot,UNDershoot,TPEXcursion,APOWer,PPPower'
station.app.write(':TPRogram:SETup:INSTruments') # clears generic instrument
station.app.write(':TPRogram:SETup:MEASurements') # Clears all measurements selections for next add
station.app.write(':TPRogram:SETup:MEASurements ' + measurements)
station.app.write(':TPRogram:SETup:EDIMage:UWBackground ON')
station.app.write(':TPRogram:SETup:EDIMage:FTYPe JPG')
def set_meas_configuration(station):
""" Specifies measurement units. """
station.app.write(':TPRogram:SETup:PUNits WATT')
station.app.write(':TPRogram:SETup:ERUNits DECibel')
station.app.write(':TPRogram:SETup:LDEFinition RLMA120')
station.app.write(':TPRogram:SETup:TRANsition SLOWest')
station.app.write(':TPRogram:SETup:THRatio 1.0E-2')
def configure_test_program(station):
""" Configures test station's test program. """
station.app.write(':TPRogram:REMove:All') # removes all current test program lines
set_fixture_lane(station)
set_waveform(station)
set_presets(station)
set_standard_measurements(station)
set_meas_configuration(station)
station.app.write(':TPRogram:SETup:ADD')
station.app.write('*CLS')
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(result, 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 = result['Value']
if 'TDECQ' in result['Name']:
measurement += ' dB'
elif 'Ceq' in result['Name']:
measurement += ' dB'
elif 'Outer OMA' in result['Name']:
measurement += ' ' + punits
elif 'Outer ER' in result['Name']:
measurement += ' ' + erunits
elif 'RLM (802.3 A_120D)' in result['Name']:
measurement += ''
elif 'Level' in result['Name']:
measurement += ' ' + punits
elif 'Trans. Time' in result['Name']:
measurement += ' ps'
elif 'Overshoot' in result['Name']:
measurement += ' %'
elif 'Undershoot' in result['Name']:
measurement += ' %'
elif 'Power Excursion' in result['Name']:
measurement += ' ' + punits
elif 'Average Power' in result['Name']:
measurement += ' ' + punits
elif 'Pk-Pk Power' in result['Name']:
measurement += ' ' + punits
return measurement
def display_job_results(station):
""" Displays the measurement results for a Session.
jobid_results = 'Fixture=DUT Fixture 1,Lane=Lane 4;Name=Average Power,Value=2.5000E-4,Status=Horrible!;...'
"""
print('\nStation ' + station.number + ' measurements:')
for job in station.job_ids:
jobid_results = station.app.query(':JOBS:RESults? ' + job)
results_list = jobid_results.split(';')
preamble = results_list.pop(0)
fixture, lane = preamble.split(',')
fixture = fixture.split('=')[1]
lane = lane.split('=')[1]
punits, erunits = get_jobid_measurement_units(station, job)
print('\n\tJob: ' + job + ',\tFixture: "' + fixture + '"\tLane: "', lane + '"')
# Converts list of measurement results to a list of single meas result Dictionaries.
# [{'Name': 'string','Value': 'string, 'Status':, 'string}, ...]
for i in range(0, len(results_list)):
temp_lst = results_list[i].split(',') # Results list of one measurement
for n in range(0, len(temp_lst)):
temp_lst[n] = temp_lst[n].split('=') # List of Name, Value, and Status results
results_list[i] = dict(temp_lst) # create dictionary with Name, Value, and Status keys
print()
aengine = station.app.query(':JOBS:RESults:INFO:AENGine? ' + job)
amodule = station.app.query(':JOBS:RESults:INFO:AMODule? ' + job)
for result in results_list:
meas_with_units = add_measurement_units_to_result(result, punits, erunits)
if 'Correct' in result['Status']:
print('\t\t' + fixture + ', ' + lane + ': ' + result['Name'] + ': ' + meas_with_units)
else:
print('\t\t' + fixture + ', ' + lane + ': "' + result['Status'] + '"')
print('\t\t\tAcquisition Engine: ' + aengine + ' with ' + 'Acquisition Module: ' + amodule)
def save_job_results(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)
tst_station.app.write(':SYSTem:DEFault')
tst_station.app.query('*OPC?')
tst_station.app.timeout = 60000 # 1 minutes
configure_test_program(tst_station)
run_again = True
while run_again:
tst_station.job_ids = run_station(tst_station) # returns new Job IDs
monitor_station(tst_station)
display_job_results(tst_station)
save_job_results(tst_station)
if rerun_test_plan():
remove_test_results(tst_station)
else:
run_again = False
close_station(tst_station)
,