forked from M-Labs/thermostat
refactor with classes
This commit is contained in:
parent
eb6ab2a222
commit
474c80722e
131
pytec/tecQT.py
131
pytec/tecQT.py
@ -6,20 +6,43 @@ from pytec.client import Client
|
||||
rec_len = 1000
|
||||
refresh_period = 20
|
||||
|
||||
channel_data = [{
|
||||
'adc': np.zeros(rec_len),
|
||||
'sens': np.zeros(rec_len),
|
||||
'temperature': np.zeros(rec_len),
|
||||
'i_set': np.zeros(rec_len),
|
||||
'pid_output': np.zeros(rec_len),
|
||||
'vref': np.zeros(rec_len),
|
||||
'dac_value': np.zeros(rec_len),
|
||||
'dac_feedback': np.zeros(rec_len),
|
||||
'i_tec': np.zeros(rec_len),
|
||||
'tec_i': np.zeros(rec_len),
|
||||
'tec_u_meas': np.zeros(rec_len),
|
||||
'interval': np.zeros(rec_len),
|
||||
} for _ in range(2)]
|
||||
class Curves:
|
||||
def __init__(self, legend: str, key: str, channel: int, color: str, buffer_len: int, period: int):
|
||||
self.curveItem = pg.PlotCurveItem(pen=({'color': color, 'width': 1}))
|
||||
self.legendStr = legend
|
||||
self.keyStr = key
|
||||
self.channel = channel
|
||||
self.data_buf = np.zeros(buffer_len)
|
||||
self.time_stamp = np.zeros(buffer_len)
|
||||
self.buffLen = buffer_len
|
||||
self.period = period
|
||||
|
||||
def update(self, tec_data, cnt):
|
||||
if cnt == 0:
|
||||
np.copyto(self.data_buf, np.full(self.buffLen, tec_data[self.channel][self.keyStr]))
|
||||
else:
|
||||
self.data_buf[:-1] = self.data_buf[1:]
|
||||
self.data_buf[-1] = tec_data[self.channel][self.keyStr]
|
||||
self.time_stamp[:-1] = self.time_stamp[1:]
|
||||
self.time_stamp[-1] = cnt * self.period / 1000
|
||||
self.curveItem.setData(x = self.time_stamp, y = self.data_buf)
|
||||
|
||||
|
||||
class Graph:
|
||||
def __init__(self, parent: pg.LayoutWidget, title: str, row: int, col: int, curves: list[Curves]):
|
||||
self.plotItem = pg.PlotWidget(title=title)
|
||||
self.legendItem = pg.LegendItem(offset=(75, 30), brush=(50,50,200,150))
|
||||
self.legendItem.setParentItem(self.plotItem.getPlotItem())
|
||||
parent.addWidget(self.plotItem, row, col)
|
||||
self.curves = curves
|
||||
for curve in self.curves:
|
||||
self.plotItem.addItem(curve.curveItem)
|
||||
self.legendItem.addItem(curve.curveItem, curve.legendStr)
|
||||
|
||||
def update(self, tec_data, cnt):
|
||||
for curve in self.curves:
|
||||
curve.update(tec_data, cnt)
|
||||
self.plotItem.setRange(xRange=[(cnt - self.curves[0].buffLen) * self.curves[0].period / 1000, cnt * self.curves[0].period / 1000])
|
||||
|
||||
tec = Client()
|
||||
|
||||
@ -36,74 +59,26 @@ cw.setLayout(l)
|
||||
|
||||
pg.setConfigOptions(antialias=True)
|
||||
|
||||
temp0plot= pg.PlotWidget(title='Channel 0 Temperature')
|
||||
layout.addWidget(temp0plot, 1, 1)
|
||||
temp1plot = pg.PlotWidget(title='Channel 1 Temperature')
|
||||
layout.addWidget(temp1plot, 2, 1)
|
||||
current0plot = pg.PlotWidget(title='Channel 0 Current')
|
||||
layout.addWidget(current0plot, 1, 2)
|
||||
current1plot = pg.PlotWidget(title='Channel 1 Current')
|
||||
layout.addWidget(current1plot, 2, 2)
|
||||
|
||||
temp0curve = pg.PlotCurveItem(pen=({'color': 'r', 'width': 1}))
|
||||
temp1curve = pg.PlotCurveItem(pen=({'color': 'r', 'width': 1}))
|
||||
tecI0curve = pg.PlotCurveItem(pen=({'color': 'r', 'width': 1}))
|
||||
tecI1curve = pg.PlotCurveItem(pen=({'color': 'r', 'width': 1}))
|
||||
Iset0curve = pg.PlotCurveItem(pen=({'color': 'g', 'width': 1}))
|
||||
Iset1curve = pg.PlotCurveItem(pen=({'color': 'g', 'width': 1}))
|
||||
temp0plot.addItem(temp0curve)
|
||||
temp1plot.addItem(temp1curve)
|
||||
current0plot.addItem(tecI0curve)
|
||||
current0plot.addItem(Iset0curve)
|
||||
current1plot.addItem(tecI1curve)
|
||||
current1plot.addItem(Iset1curve)
|
||||
|
||||
temp0legend = temp0plot.getPlotItem().addLegend(brush=(50,50,200,150))
|
||||
temp0legend.addItem(temp0curve, 'feedback')
|
||||
temp1legend = temp1plot.getPlotItem().addLegend(brush=(50,50,200,150))
|
||||
temp1legend.addItem(temp0curve, 'feedback')
|
||||
current0legend = current0plot.getPlotItem().addLegend(brush=(50,50,200,150))
|
||||
current0legend.addItem(tecI0curve, 'feedback')
|
||||
current0legend.addItem(Iset0curve, 'setpoint')
|
||||
current1legend = current1plot.getPlotItem().addLegend(brush=(50,50,200,150))
|
||||
current1legend.addItem(tecI1curve, 'feedback')
|
||||
current1legend.addItem(Iset1curve, 'setpoint')
|
||||
ch0tempGraph = Graph(layout, 'Channel 0 Termperature', 1, 1, [Curves('Feedback', 'temperature', 0, 'r', rec_len, refresh_period)])
|
||||
ch1tempGraph = Graph(layout, 'Channel 1 Termperature', 2, 1, [Curves('Feedback', 'temperature', 1, 'r', rec_len, refresh_period)])
|
||||
ch0currentGraph = Graph(layout, 'Channel 0 Current', 1, 2, [Curves('Feedback', 'tec_i', 0, 'r', rec_len, refresh_period),
|
||||
Curves('Setpoint', 'i_set', 0, 'g', rec_len, refresh_period)])
|
||||
ch1currentGraph = Graph(layout, 'Channel 1 Current', 2, 2, [Curves('Feedback', 'tec_i', 1, 'r', rec_len, refresh_period),
|
||||
Curves('Setpoint', 'i_set', 1, 'g', rec_len, refresh_period)])
|
||||
|
||||
cnt = 0
|
||||
time_stamp = np.zeros(rec_len)
|
||||
def update(n):
|
||||
for data in tec.report_mode():
|
||||
ch = data[n]
|
||||
for tag, seq in channel_data[n].items():
|
||||
if tag in ch:
|
||||
v = ch[tag]
|
||||
if type(v) is float:
|
||||
if cnt == 0:
|
||||
np.copyto(seq, np.full(rec_len, v))
|
||||
else:
|
||||
seq[:-1] = seq[1:]
|
||||
seq[-1] = v
|
||||
if quit:
|
||||
break
|
||||
return
|
||||
|
||||
def updateData():
|
||||
global cnt
|
||||
update(0)
|
||||
update(1)
|
||||
cnt += 1
|
||||
time_stamp[:-1] = time_stamp[1:]
|
||||
time_stamp[-1] = cnt * refresh_period / 1000
|
||||
temp0plot.setRange(xRange=[(cnt - rec_len) * refresh_period / 1000, cnt * refresh_period / 1000])
|
||||
temp1plot.setRange(xRange=[(cnt - rec_len) * refresh_period / 1000, cnt * refresh_period / 1000])
|
||||
current0plot.setRange(xRange=[(cnt - rec_len) * refresh_period / 1000, cnt * refresh_period / 1000])
|
||||
current1plot.setRange(xRange=[(cnt - rec_len) * refresh_period / 1000, cnt * refresh_period / 1000])
|
||||
temp0curve.setData(x = time_stamp, y = channel_data[0]['temperature'])
|
||||
temp1curve.setData(x = time_stamp, y = channel_data[1]['temperature'])
|
||||
tecI0curve.setData(x = time_stamp, y = channel_data[0]['tec_i'])
|
||||
tecI1curve.setData(x = time_stamp, y = channel_data[1]['tec_i'])
|
||||
Iset0curve.setData(x = time_stamp, y = channel_data[0]['i_set'])
|
||||
Iset1curve.setData(x = time_stamp, y = channel_data[1]['i_set'])
|
||||
for data in tec.report_mode():
|
||||
|
||||
ch0tempGraph.update(data, cnt)
|
||||
ch1tempGraph.update(data, cnt)
|
||||
ch0currentGraph.update(data, cnt)
|
||||
ch1currentGraph.update(data, cnt)
|
||||
|
||||
if quit:
|
||||
break
|
||||
cnt += 1
|
||||
|
||||
|
||||
## Start a timer to rapidly update the plot in pw
|
||||
|
||||
Loading…
Reference in New Issue
Block a user