Example of Instrument Driver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
import sys
import json
import pyvisa

DONE = 'DONE'
InputNames = ['In']
Model = 'UNKNOWN'
Serial = 'UNKNOWN'

# FlexOTO is listening to the standard output
def send_to_FlexOTO(message: str):
  print(message)


# Check the error queue
def check_error(inst: pyvisa.Resource) -> tuple[int, str]:
   error = inst.query('SYST:ERR?').strip()
   sections = error.split(',', 1)
   return (int(sections[0]), sections[1].strip('"'))


# Open connection to instrument
def connect(visaAddress: str) -> pyvisa.Resource:
   try:
      rm = pyvisa.ResourceManager()
      return rm.open_resource(visaAddress)
   except:
      send_to_FlexOTO(f'Failed to connect to "{visaAddress}"')
      return None


# Check if this is a supported instrument.
def validate(inst: pyvisa.Resource) -> bool:

   if inst is None: return False

   # Make sure we are connected to a supported Lightwave Mainframe.
   mainframeIdn = inst.query('*IDN?')
   sections = mainframeIdn.split(',')
   if len(sections) < 3:
      send_to_FlexOTO('Please connect to a 816x mainframe with a power meter in Slot 1.')
      return False

   mfManufacturer = sections[0].upper().strip()
   if not (mfManufacturer.startswith('KEYSIGHT') or
           mfManufacturer.startswith('AGILENT') or
           mfManufacturer.startswith('HEWLETT') or
           mfManufacturer.startswith('HP')):
      send_to_FlexOTO('Please connect to a 816x mainframe with a power meter in Slot 1.')
      return False

   mfModel = sections[1].strip()
   if not mfModel.startswith('816'):
      send_to_FlexOTO('Please connect to a 816x mainframe with a power meter in Slot 1.')
      return False

   # Found an 816x mainframe (e.g. 8163B)
   # Make sure a supported power meter is in Slot 1.
   moduleIdn = inst.query('SLOT1:IDN?')
   sections = moduleIdn.split(',')
   if len(sections) < 3:
      send_to_FlexOTO('Please install the power meter module in Slot 1')
      return False

   manufacturer = sections[0].upper().strip()
   if manufacturer.startswith('KEYSIGHT'):
      manufacturer = 'Keysight'
   elif manufacturer.startswith('AGILENT'):
      manufacturer = 'Agilent'
   elif manufacturer.startswith('HEWLETT') or manufacturer.startswith('HP'):
      manufacturer = 'HP'
   else:
      send_to_FlexOTO('Unsupported manufacturer: ' + manufacturer)
      return False

   model = sections[1].strip()
   if not model.startswith('8163'):
      send_to_FlexOTO('Unrecognized power meter model: ' + model)
      return False

   # Found an 8163x power meter (e.g. 81634A)
   # Save the model and serial numbers for later.
   global Model, Serial
   Model = manufacturer + " " + model
   Serial = sections[2].strip()[-5:] # Get last 5 of serial number
   return True


# Do the initial setup of the instrument
def initialize(inst: pyvisa.Resource):

   # Set timeout to 10 sec. This should work for all commands except zeroing.
   inst.timeout = 10000

   # Make sure that the reference is not used.
   inst.write('SENS1:CHAN1:POW:REF:STATE 0')

   # Turn auto range on.
   inst.write('SENS1:CHAN1:POW:RANGE:AUTO 1')

   # Change the power unit to Watt.
   inst.write('SENS1:CHAN1:POW:UNIT W')

   # Set the averaging time for measuring to 0.5s.
   inst.write('SENS1:CHAN1:POW:ATIME 0.5')
   
   # Turn continuous measuring off.
   inst.write('INIT1:CHAN1:CONT 0')


# Blocks until a command comes in from FlexOTO,
# and then extracts the command and arguments strings.
def wait_for_input() -> tuple[str, list[str]]:

   # Commands come from the standard input.
   rawInput = input()
   sections = rawInput.split('"')

   items = []
   for s in sections:
      s = s.strip()
      if s: items.append(s)

   command = items.pop(0)
   args = items

   return (command, args)


# Sends the JSON description of the instrument to FlexOTO.
def get_description():

   # Use the model and serial numbers determined earlier
   desc = json.dumps({ 'ModelNumber': Model,
                       'SerialNumber': Serial,
                       'Inputs': InputNames,
                       'MeasurementTimeoutSeconds': 10 })
   
   send_to_FlexOTO(desc)
   send_to_FlexOTO(DONE)


# Measure the active inputs and send the results to FlexOTO.
def measure(inst: pyvisa.Resource, activeInputs: list[str]):
   
   # Clear error queue
   inst.write('*CLS')

   measList = []
   for inputName in activeInputs:
      inputNum = InputNames.index(inputName) + 1

      # Make an average power measurement on this channel.
      avgPower = float(inst.query('READ1:CHAN{0}:POW?'.format(inputNum)))

      # Start a dictionary to describe a measurement of Average Power on this input.
      measurement = { 'Name': 'Average Power', 'Input': inputName }

      (errorCode, errorMsg) = check_error(inst)

      if errorCode == 0:
         measurement['Result'] = avgPower
         measurement['FormattedResult'] = '{0:.2f} \u03BCW'.format(avgPower * 1e6) # Format in uW
      else:
         # Report the error
         measurement['Result'] = float("NaN") # NaN ("not a number") indicates an invalid result
         measurement['FormattedResult'] = errorMsg

      # Add the result to the measurements list.
      measList.append(measurement)

   measJson = json.dumps({'Measurements': measList}, allow_nan=True)

   send_to_FlexOTO(measJson)
   send_to_FlexOTO(DONE)



# Program begins here
if len(sys.argv) < 2:
   send_to_FlexOTO("Please provide the instrument's VISA address (TCPIP0::HOSTNAME::inst0::INSTR) in the Command Line Arguments.")
else:

   # Get the instrument's VISA address from the command line args.
   visaAddress = sys.argv[1]
   inst = connect(visaAddress)

   # Check if this is a valid instrument.
   if validate(inst):
   
      # Do initial configuration of instrument setup.
      initialize(inst)

# Connection and initial setup is done.
send_to_FlexOTO(DONE)

########## Main loop ##########

# Loop until FlexOTO sends us 'exit'.
exit = False
while not exit:
   (command, args) = wait_for_input()
 
   if command == 'exit':
      exit = True
 
   elif command == 'get_description':
      get_description()
 
   elif command == 'measure':
      measure(inst, args)
      

# Exiting...