bdubois / n523x_utils

Download as
Browse Code ยป

Commit 9245ec8da86ed48de1a7bb70cd322f2abfca8301

Authored by bdubois
0 parents
Exists in master

first commit

Showing 3 changed files with 650 additions and 0 deletions Inline Diff

File was created 1 Handle PNA-X, N522xA or N523x device from Keysight. Tested on N5234 device.
n523x_utils/extract_peak.py
Diff comments   View file @ 9245ec8
File was created 1 #!/usr/bin/python3
2 # -*- coding: utf-8 -*-
3
4 """package n523xA_utils
5 author Benoit Dubois
6 copyright FEMTO ENGINEERING
7 license GPL v3.0+
8 brief Acquire data trace from N5234A or N5230A device.
9 Try to detrend trace in order to extract mode shape.
10 """
11
12 # Ctrl-c closes the application
13 import signal
14 signal.signal(signal.SIGINT, signal.SIG_DFL)
15
16 import sys
17 import os.path as path
18 import logging
19 from pyqtgraph.parametertree import Parameter, ParameterTree
20 import pyqtgraph as pg
21 import numpy as np
22 import numpy.polynomial.polynomial as poly
23 import scipy.signal as scs
24 from PyQt4.QtCore import pyqtSlot, pyqtSignal, QDir, QFileInfo
25 from PyQt4.QtGui import QApplication, QMainWindow, QWidget, QVBoxLayout, \
26 QHBoxLayout, QMessageBox, QFileDialog
27
28 import n523x as vna
29
30 CONSOLE_LOG_LEVEL = logging.DEBUG
31 FILE_LOG_LEVEL = logging.WARNING
32
33 APP_NAME = "ExtractPeak"
34
35 #===============================================================================
36 class MyVna(vna.N523x):
37
38 def connect(self, try_=3):
39 """Overloaded vna.Vna method beacause, VNA seems to refuse connection
40 without reasons.
41 :param try_: number of connection attempt (int).
42 """
43 for i in range(try_):
44 if super().connect() is True:
45 break
46
47 #===============================================================================
48 PARAMS = [
49 {'name': 'Load data', 'type': 'group', 'children': [
50 {'name': 'VNA', 'type': 'group', 'children': [
51 {'name': 'IP', 'type': 'str', 'value': vna.DEFAULT_IP},
52 {'name': 'Port', 'type': 'int', 'value': vna.DEFAULT_PORT},
53 {'name': 'Acquisition', 'type': 'action'},
54 ]},
55 {'name': 'File', 'type': 'group', 'children': [
56 {'name': 'Filename', 'type': 'str'},
57 {'name': 'Open', 'type': 'action'},
58 ]},
59 ]},
60 {'name': 'Fit', 'type': 'group', 'children': [
61 {'name': 'Order', 'type': 'int', 'value': 3, 'limits': (1, 15)},
62 {'name': 'Filtering', 'type': 'bool', 'value': False},
63 {'name': 'Run fit', 'type': 'action'},
64 ]},
65 {'name': 'Plot', 'type': 'group', 'children': []},
66 ]
67
68 class PeakUi(QMainWindow):
69 """Ui of extract peak application.
70 """
71
72 def __init__(self):
73 """Constructor.
74 :returns: None
75 """
76 super().__init__()
77 self.setWindowTitle("Fit Peak")
78 self.setCentralWidget(self._central_widget())
79 self.region = pg.LinearRegionItem()
80 self.region.setZValue(10)
81 # Add the LinearRegionItem to the ViewBox, but tell the ViewBox to
82 # exclude this item when doing auto-range calculations.
83 self.mplot1.addItem(self.region, ignoreBounds=True)
84
85 def _central_widget(self):
86 """Generates central widget.
87 :returns: central widget of UI (QWidget)
88 """
89 self.p = Parameter.create(name='params', type='group', children=PARAMS)
90 self.ptree = ParameterTree()
91 self.ptree.setParameters(self.p, showTop=False)
92 self.mplot1 = pg.PlotWidget()
93 self.mplot2 = pg.PlotWidget()
94 plot_lay = QVBoxLayout()
95 plot_lay.addWidget(self.mplot1)
96 plot_lay.addWidget(self.mplot2)
97 main_lay = QHBoxLayout()
98 main_lay.addWidget(self.ptree)
99 main_lay.addLayout(plot_lay)
100 main_lay.setStretchFactor(plot_lay, 2)
101 central_widget = QWidget()
102 central_widget.setLayout(main_lay)
103 return central_widget
104
105 #===============================================================================
106 class PeakApp(QApplication):
107 """Peak application.
108 """
109
110 # Emit data key value
111 acquire_done = pyqtSignal()
112 fit_done = pyqtSignal(object, object)
113
114 def __init__(self, args):
115 """Constructor.
116 :returns: None
117 """
118 super().__init__(args)
119 self._ui = PeakUi()
120 self._data = None
121 self._cdata_name = None
122 self._dplot = None # Data plot
123 self._fdplot = None # Fitted data plot
124 self._pplot = None # (Extracted) Peak plot
125 self._ui.p.param(
126 'Load data', 'VNA', 'Acquisition').sigActivated.connect(
127 self.acquire_data_from_device)
128 self._ui.p.param('Load data', 'File', 'Open').sigActivated.connect(
129 self.acquire_data_from_file)
130 self._ui.p.param('Fit', 'Run fit').sigActivated.connect(
131 lambda: self.fit_peak(self._ui.p.param('Fit', 'Order').value(),
132 self._ui.p.param('Fit', 'Filtering').value()))
133 self.acquire_done.connect(self.handle_acq_data)
134 self.fit_done.connect(self.display_fit)
135 self._ui.show()
136
137 @pyqtSlot()
138 def acquire_data_from_device(self):
139 """Acquire data process.
140 :returns: None
141 """
142 ip = self._ui.p.param('Load data', 'VNA', 'IP').value()
143 port = self._ui.p.param('Load data', 'VNA', 'Port').value()
144 # Acquisition itself
145 dev = MyVna(ip, port)
146 dev.connect()
147 dev.write("FORMAT:DATA REAL,64")
148 try:
149 datas = dev.get_measurements()
150 except Exception as ex:
151 logging.error("Problem during acquisition: %s", str(ex))
152 QMessageBox.warning(self._ui, "Acquisition problem",
153 "Problem during acquisition: {}".format(ex),
154 QMessageBox.Ok)
155 return
156 self._data = {dev.measurement_number_to_name(idx+1):
157 data for idx, data in enumerate(datas)}
158 self.acquire_done.emit()
159
160 @pyqtSlot()
161 def acquire_data_from_file(self):
162 filename = QFileDialog().getOpenFileName(
163 parent=None,
164 caption="Choose .s2p file to load",
165 directory=QDir.currentPath(),
166 filter="s2p files (*.s2p);;Any files (*)")
167 if filename == '':
168 return
169 try:
170 data = np.transpose(np.loadtxt(filename, comments=('!', '#')))
171 except Exception as ex:
172 logging.error("Problem when reading file: %s", str(ex))
173 QMessageBox.warning(self._ui, "Acquisition problem",
174 "Problem when reading file: {}".format(ex),
175 QMessageBox.Ok)
176 return
177 self._data = {QFileInfo(filename).baseName(): data}
178 self.acquire_done.emit()
179
180 @pyqtSlot()
181 def handle_acq_data(self):
182 self._ui.p.param('Plot').clearChildren()
183 for name_, data in self._data.items():
184 self._ui.p.param('Plot').addChild(
185 Parameter.create(name=name_, type='action'))
186 for child in self._ui.p.param('Plot').children():
187 child.sigActivated.connect(self.set_cdata)
188 self.set_cdata(self._ui.p.param('Plot').children()[0])
189
190 @pyqtSlot(object)
191 def set_cdata(self, param):
192 """
193 :param param: Parameter object
194 """
195 self._cdata_name = param.name()
196 self.display_data()
197
198 @pyqtSlot()
199 def display_data(self):
200 if self._dplot is not None:
201 self._ui.mplot1.removeItem(self._dplot)
202 if self._fdplot is not None:
203 self._ui.mplot1.removeItem(self._fdplot)
204 if self._pplot is not None:
205 self._ui.mplot2.removeItem(self._pplot)
206 self._dplot = self._ui.mplot1.plot(
207 self._data[self._cdata_name][0, :],
208 self._data[self._cdata_name][1, :], pen="w")
209 self._ui.region.setRegion([self._data[self._cdata_name][0, 0],
210 self._data[self._cdata_name][0, -1]])
211
212 @pyqtSlot(int, bool)
213 def fit_peak(self, order, filter_=False):
214 name_ = self._cdata_name
215 x_min, x_max = self._ui.region.getRegion()
216 x_min_id = np.where(self._data[name_][0, :] >= x_min)[0][0]
217 x_max_id = np.where(self._data[name_][0, :] <= x_max)[0][-1]
218 x_n = np.concatenate([self._data[name_][0, 1:x_min_id],
219 self._data[name_][0, x_max_id:-1]])
220 # Suppress mean to get better fitting:
221 x_n0 = x_n - self._data[name_][0, :].mean()
222 y_n = np.concatenate([self._data[name_][1, 1:x_min_id],
223 self._data[name_][1, x_max_id:-1]])
224 try:
225 coefs = poly.polyfit(x_n0, y_n, order)
226 except TypeError: # when user has not selected region to exclude
227 QMessageBox.information(self._ui,
228 "ROI omited",
229 "Select region to exclude before fit")
230 return
231 xfit = self._data[name_][0, :] - self._data[name_][0, :].mean()
232 yfit = poly.polyval(xfit, coefs)
233 res = self._data[name_][1, :] - yfit
234 if filter_ is True:
235 b, a = scs.butter(3, 0.005)
236 res = scs.filtfilt(b, a, res, padlen=150)
237 self.fit_done.emit(yfit, res)
238
239 @pyqtSlot(object, object)
240 def display_fit(self, yfit, res):
241 name_ = self._cdata_name
242 if self._fdplot is not None:
243 self._ui.mplot1.removeItem(self._fdplot)
244 if self._pplot is not None:
245 self._ui.mplot2.removeItem(self._pplot)
246 self._fdplot = self._ui.mplot1.plot(
247 self._data[name_][0, :], yfit, pen='g')
248 self._pplot = self._ui.mplot2.plot(
249 self._data[name_][0, :]-self._data[name_][0, :].mean(),
250 res, pen='g')
251
252 #==============================================================================
253 def configure_logging():
254 """Configures logs.
255 """
256 home = path.expanduser("~")
257 log_file = "." + APP_NAME + ".log"
258 abs_log_file = path.join(home, log_file)
259 date_fmt = "%d/%m/%Y %H:%M:%S"
260 log_format = "%(asctime)s %(levelname) -8s %(filename)s " + \
261 " %(funcName)s (%(lineno)d): %(message)s"
n523x_utils/n523x.py
Diff comments   View file @ 9245ec8
File was created 1 # -*- coding: utf-8 -*-
2
3 """package n523x_utils
4 author Benoit Dubois
5 copyright FEMTO ENGINEERING
6 license GPL v3.0+
7 brief Handle PNA-X, N522xA or N523x device from Keysight.
8 Tested on N5234 device.
9 """
10
11 import time
12 import socket
13 import struct
14 import logging
15 import numpy as np
16
17 DEFAULT_IP = "192.168.0.11"
18 DEFAULT_PORT = 5025
19
20 #==============================================================================
21 class N523x(object):
22 """Handle PNA-X, N522xA or N523x device from Keysight.
23 """
24
25 IFBW = (1, 2, 3, 5, 7, 10, 15, 20, 30, 50, 70, 100, 150, 200, 300, 500, 700,
26 1e3, 1.5e3, 2e3, 3e3, 5e3, 7e3, 10e3, 15e3, 20e3, 30e3, 50e3, 70e3,
27 100e3, 150e3, 200e3, 280e3, 360e3, 600e3)
28
29 def __init__(self, ip, port=DEFAULT_PORT, timeout=1.0):
30 """Constructor.
31 :param ip: IP address of device (str)
32 :param port: Ethernet port of device (int)
33 :param timeout: Timeout on connection instance (float)
34 :returns: None
35 """
36 self.ip = ip
37 self.port = port
38 self._timeout = timeout
39 self._sock = None
40
41 def connect(self):
42 """Connect to device.
43 """
44 try:
45 self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
46 self._sock.settimeout(self._timeout)
47 self._sock.connect((self.ip, self.port))
48 except ValueError as ex:
49 logging.error("Connection parameters out of range: %s", str(ex))
50 return False
51 except socket.timeout:
52 logging.error("Timeout on connection")
53 return False
54 except Exception as ex:
55 logging.error("Unexpected exception during connection with " + \
56 "VNA: %s", str(ex))
57 return False
58 else:
59 logging.debug("Connected to VNA")
60 return True
61
62 def write(self, data):
63 """"Ethernet writing process.
64 :param data: data writes to device (str)
65 :returns: None
66 """
67 try:
68 self._sock.send((data + "\n").encode('utf-8'))
69 except socket.timeout:
70 logging.error("Timeout")
71 except Exception as ex:
72 logging.error(str(ex))
73 logging.debug("write " + data.strip('\n'))
74
75 def read(self, length=100):
76 """Read process.
77 :param length: length of message to read (int)
78 :returns: Message reads from device (str)
79 """
80 try:
81 retval = self._sock.recv(length).decode('utf-8').strip('\n')
82 except socket.timeout:
83 logging.error("Timeout")
84 return ''
85 except Exception as ex:
86 logging.error(str(ex))
87 raise ex
88 logging.debug("read: " + retval)
89 return retval
90
91 def raw_read(self, length=512):
92 """Raw read process.
93 :param length: length of message to read (int)
94 :returns: Message reads from device (bytes)
95 """
96 try:
97 data = self._sock.recv(length)
98 except socket.timeout:
99 logging.error("Timeout")
100 return bytes()
101 except Exception as ex:
102 logging.error(str(ex))
103 raise ex
104 return data
105
106 def bin_read(self):
107 """Read binary data then decode them to ascii.
108 The reading process is specific to the transfert of binary data
109 with these VNA devices: <header><data><EOT>, with:
110 - <header>: #|lenght of bytes_count (one byte)|bytes_count
111 - <data>: "REAL,64" (float64) binary data
112 - <EOT>: '\n' character.
113 Note: The data transfert format must be selected to "REAL,64"" before
114 using this method
115 """
116 header_max_length = 11
117 raw_data = self.raw_read(header_max_length)
118 if raw_data.find(b'#') != 0:
119 logging.error("Data header not valid")
120 return
121 byte_count_nb = int(raw_data[1:2])
122 byte_count = int(raw_data[2:2+byte_count_nb])
123 # Note : Read 'byte_count' bytes but only
124 # 2 + byte_count_nb + byte_count - header_max_length
125 # needs to be readen.
126 # This tip can be used because EOF ('\n') is transmited at the end of
127 # the message and thus stop reception of data.
128 while len(raw_data) < byte_count:
129 raw_data += self.raw_read(byte_count)
130 nb_elem = int(byte_count / 8)
131 data = np.asarray(struct.unpack(">{:d}d".format(nb_elem),
132 raw_data[2+byte_count_nb:-1]))
133 return data
134
135 def query(self, msg, length=100):
136 """Basic query process: write query then read response.
137 """
138 self.write(msg)
139 return self.read(length)
140
141 def reset(self):
142 """Reset device.
143 """
144 self.write("*RST")
145
146 @property
147 def id(self):
148 """Return ID of device.
149 """
150 return self.query("*IDN?")
151
152 def get_span(self, cnum=1):
153 return self.query("SENS{}:FOM:RANG:SEGM:FREQ:SPAN?".format(cnum))
154
155 def set_span(self, value, cnum=1):
156 self.write("SENS{}:FREQ:SPAN {}".format(cnum, value))
157
158 def get_start(self, cnum=1):
159 return self.query("SENS{}:FREQ:START?".format(cnum))
160
161 def set_start(self, value, cnum=1):
162 self.write("SENS{}:FREQ:START {}".format(cnum, value))
163
164 def get_stop(self, cnum=1):
165 return self.query("SENS{}:FREQ:STOP?".format(cnum))
166
167 def set_stop(self, value, cnum=1):
168 self.write("SENS{}:FREQ:STOP {}".format(cnum, value))
169
170 def get_center_freq(self, cnum=1):
171 return self.query("SENS{}:FREQ:CENT?".format(cnum))
172
173 def set_center_freq(self, value, cnum=1):
174 self.write("SENS{}:FREQ:CENT {}".format(cnum, value))
175
176 def get_points(self, cnum=1):
177 return self.query("SENS{}:SWE:POINTS?".format(cnum))
178
179 def set_points(self, value, cnum=1):
180 self.write("SENS{}:SWE:POINTS {}".format(cnum, value))
181
182 def get_sweep_type(self, cnum=1):
183 return self.query("SENS{}:SWE:TYPE?".format(cnum))
184
185 def set_sweep_type(self, value, cnum=1):
186 self.write("SENS{}:SWE:TYPE {}".format(cnum, value))
187
188 @staticmethod
189 def read_s2p(filename):
190 return np.loadtxt(filename, comments=['!', '#'])
191
192 def get_window_numbers(self):
193 """Return the number of existing windows.
194 """
195 data = self.query("DISP:CAT?")
196 if data is None:
197 return []
198 return [int(x) for x in data.replace('\"', '').split(',')]
199
200 def get_measurement_catalog(self, channel=''):
201 """Returns ALL measurement numbers, or measurement numbers
202 from a specified channel
203 :param channel: Channel number to catalog. If not specified,
204 all measurement numbers are returned.
205 :returns: ALL measurement numbers, or measurement numbers
206 from a specified channel
207 """
208 data = self.query("SYST:MEAS:CAT? {}".format(channel))
209 if data is None:
210 return []
211 return [int(x) for x in data.replace('\"', '').split(',')]
212
213 def measurement_number_to_trace(self, nb=None):
214 """Returns the trace number of the specified measurement number.
215 Trace numbers restart for each window while measurement numbers are
216 always unique.
217 :param n: Measurement number for which to return the trace number.
218 If unspecified, value is set to 1.
219 """
220 return self.query("SYST:MEAS{}:TRAC?".format(nb))
221
222 def measurement_number_to_name(self, nb=None):
223 """Returns the name of the specified measurement.
224 :param n: Measurement number for which to return the measurement name.
225 If unspecified, value is set to 1.
226 """
227 return self.query("SYST:MEAS{}:NAME?".format(nb)).replace('\"', '')
228
229 def set_measurement(self, name, fast=True):
230 """ Sets the selected measurement. Most CALC: commands require that
231 this command be sent before a setting change is made. One measurement
232 on each channel can be selected at the same time.
233 :param name: Name of the measurement. CASE-SENSITIVE. Do NOT include
234 the parameter name that is returned with Calc:Par:Cat?
235 :param fast: Optional. The PNA display is NOT updated. Therefore,
236 do not use this argument when an operator is using the
237 PNA display. Otherwise, sending this argument results
238 in much faster sweep speeds. There is NO other reason
239 to NOT send this argument.
240 """
241 if name is None:
242 logging.error("Requiered name parameter")
243 raise ValueError("Requiered name parameter")
244 cnum = int(name[2])
245 self.write("CALC{}:PAR:SEL {}{}"
246 .format(cnum, name, ",fast" if fast is True else None))
247
248 def get_measurement(self, name):
249 """Get a data measurement.
250 Note that VNA must be configured to transfer data in float 64 format
251 before using the method.
252 :param name: Name of the measurement. CASE-SENSITIVE. Do NOT include
253 the parameter name that is returned with Calc:Par:Cat?
254 :return: Array of measurement data.
255 """
256 cnum = int(name[2])
257 self.set_measurement(name)
258 if self.get_sweep_type() != "LIN":
259 raise NotImplementedError
260 datax = np.linspace(float(self.get_start(cnum)),
261 float(self.get_stop(cnum)),
262 int(self.get_points(cnum)))
263 self.write("CALC{}:DATA? FDATA".format(cnum))
264 datay = self.bin_read()
265 retval = np.asarray([datax, datay])
266 return retval
267
268 def get_snp(self, cnum=1):
269 """Get snp data.
270 :param cnum: Channel number of the measurement. There must be a selected
271 measurement on that channel. If unspecified, <cnum> is set to 1.
272 :return: Array of data.
273 """
274 self.write("CALC{}:DATA:SNP:PORT? \"1,2\"".format(cnum))
275 return np.asarray(self.bin_read()).reshape(9, -1)
276
277 def get_measurements(self):
278 """Find current measurements, get data then prepare a list of array
279 [freq, data] for each measurement.
280 :returns: list of measurements
281 """
282 datas = []
283 meas_nb = self.get_measurement_catalog()
284 for nb in meas_nb:
285 name = self.measurement_number_to_name(nb)
286 datas.append(self.get_measurement(name))
287 return datas
288
289 def get_snps(self):
290 """Get all snp traces.
291 """
292 meas_nb = self.get_measurement_catalog()
293 datas = []
294 for measurement in meas_nb:
295 datas.append(self.get_snp(measurement))
296 return datas
297
298 def write_snps(self, filename=None):
299 """Write all snp traces.
300 """
301 if filename is None:
302 filename = time.strftime("%Y%m%d-%H%M%S", time.gmtime())
303 meas_nb = self.get_measurement_catalog()
304 for measurement in meas_nb:
305 data = np.asarray(self.get_snp(measurement)).reshape(9, -1)
306 np.savetxt(filename + '_' + str(measurement) + '.s2p',
307 data,
308 delimiter='\t',
309 header="#f(Hz)\tReal(S11)\tImag(S11)" +
310 "\tReal(S21)\tImag(S21)" +
311 "\tReal(S12)\tImag(S12)" +
312 "\tReal(S22)\tImag(S22)")
313
314
315 #==============================================================================