forked from M-Labs/kirdy
836 lines
40 KiB
Python
836 lines
40 KiB
Python
from PyQt6 import QtWidgets, QtGui, QtCore
|
|
from PyQt6.QtCore import pyqtSignal, QObject, QSignalBlocker, pyqtSlot
|
|
import pyqtgraph.parametertree.parameterTypes as pTypes
|
|
from pyqtgraph.parametertree import Parameter, ParameterTree, ParameterItem, registerParameterType
|
|
import pyqtgraph as pg
|
|
pg.setConfigOptions(antialias=True)
|
|
from pyqtgraph import mkPen
|
|
from pglive.sources.live_axis_range import LiveAxisRange
|
|
from pglive.sources.data_connector import DataConnector
|
|
from pglive.kwargs import Axis, LeadingLine
|
|
from pglive.sources.live_plot import LiveLinePlot
|
|
from pglive.sources.live_plot_widget import LivePlotWidget
|
|
from pglive.sources.live_axis import LiveAxis
|
|
import sys
|
|
import argparse
|
|
import logging
|
|
import asyncio
|
|
from driver.kirdy_async import Kirdy, StoppedConnecting
|
|
import qasync
|
|
from qasync import asyncSlot, asyncClose
|
|
from collections import deque
|
|
from datetime import datetime, timezone, timedelta
|
|
from time import time
|
|
from typing import Any, Optional, List
|
|
from ui.ui_conn_settings_form import Ui_Conn_Settings_Form
|
|
from ui.ui_kirdy_qt import Ui_MainWindow
|
|
from dateutil import tz
|
|
import math
|
|
|
|
def get_argparser():
|
|
parser = argparse.ArgumentParser(description="ARTIQ master")
|
|
|
|
parser.add_argument("--connect", default=None, action="store_true",
|
|
help="Automatically connect to the specified Thermostat in IP:port format")
|
|
parser.add_argument('IP', metavar="ip", default=None, nargs='?')
|
|
parser.add_argument('PORT', metavar="port", default=None, nargs='?')
|
|
parser.add_argument("-l", "--log", dest="logLevel", choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
|
|
help="Set the logging level")
|
|
|
|
return parser
|
|
|
|
class KirdyDataWatcher(QObject):
|
|
"""
|
|
This class provides various signals for Mainwindow to update various kinds of GUI objects
|
|
"""
|
|
connection_error_sig = pyqtSignal()
|
|
setting_update_sig = pyqtSignal(dict)
|
|
report_update_sig = pyqtSignal(dict)
|
|
|
|
def __init__(self, parent, kirdy, update_s):
|
|
self._update_s = update_s
|
|
self._kirdy = kirdy
|
|
self._watch_task = None
|
|
self._report_mode_task = None
|
|
self._poll_for_report = True
|
|
super().__init__(parent)
|
|
|
|
async def signal_emitter(self):
|
|
try:
|
|
settings_summary = await self._kirdy.device.get_settings_summary()
|
|
self.setting_update_sig.emit(settings_summary)
|
|
if self._poll_for_report:
|
|
status_report = await self._kirdy.device.get_status_report()
|
|
self.report_update_sig.emit(status_report)
|
|
# TODO: Identify the possible types of error that is connection related
|
|
except:
|
|
logging.error("Client connection error, disconnecting", exc_info=True)
|
|
self._kirdy.stop_report_mode()
|
|
self.connection_error_sig.emit()
|
|
|
|
async def run(self):
|
|
while True:
|
|
asyncio.ensure_future(self.signal_emitter())
|
|
await asyncio.sleep(self._update_s)
|
|
|
|
def start_watching(self):
|
|
self._watch_task = asyncio.create_task(self.run())
|
|
|
|
def stop_watching(self):
|
|
if self._watch_task is not None:
|
|
self._watch_task.cancel()
|
|
self._watch_task = None
|
|
|
|
async def set_report_mode(self, enabled: bool):
|
|
self._poll_for_report = not enabled
|
|
if enabled:
|
|
self._report_mode_task = asyncio.create_task(self.report_mode())
|
|
else:
|
|
self._kirdy.stop_report_mode()
|
|
if self._report_mode_task is not None:
|
|
await self._report_mode_task
|
|
self._report_mode_task = None
|
|
|
|
async def report_mode(self):
|
|
try:
|
|
async for status_report in self._kirdy.report_mode():
|
|
self.report_update_sig.emit(status_report)
|
|
except:
|
|
logging.error("Client connection error, disconnecting", exc_info=True)
|
|
self._kirdy.stop_report_mode()
|
|
self.connection_error_sig.emit()
|
|
|
|
@pyqtSlot(float)
|
|
def set_update_s(self, update_s):
|
|
self._update_s = update_s
|
|
|
|
class Graphs:
|
|
def __init__(self, ld_i_set_graph, pd_mon_pwr_graph, tec_i_graph, tec_temp_graph, max_samples=1000):
|
|
self.graphs = [ld_i_set_graph, pd_mon_pwr_graph, tec_i_graph, tec_temp_graph]
|
|
self.connectors = []
|
|
|
|
self._pd_mon_pwr_plot = LiveLinePlot(pen=pg.mkPen('r'))
|
|
self._ld_i_set_plot = LiveLinePlot(name="Set", pen=pg.mkPen('r'))
|
|
self._tec_temp_plot = LiveLinePlot(pen=pg.mkPen('r'))
|
|
self._tec_setpoint_plot = LiveLinePlot(pen=pg.mkPen('r'))
|
|
self._tec_i_target_plot = LiveLinePlot(name="Target", pen=pg.mkPen('r'))
|
|
self._tec_i_measure_plot = LiveLinePlot(name="Measure", pen=pg.mkPen('g'))
|
|
|
|
self._temp_setpoint_line = tec_temp_graph.getPlotItem().addLine(label='{value} °C', pen=pg.mkPen('g'))
|
|
# Render the temperature setpoint line on top of the temperature being plotted
|
|
self._temp_setpoint_line.setZValue(10)
|
|
self._temp_setpoint_line.setVisible(False)
|
|
|
|
def tickStrings(values: List, scale: float, spacing: float) -> List:
|
|
return [datetime.fromtimestamp(value/1000, tz=timezone.utc).strftime("%H:%M:%S") for value in values]
|
|
|
|
for graph in ld_i_set_graph, pd_mon_pwr_graph, tec_i_graph, tec_temp_graph:
|
|
time_axis = LiveAxis('bottom', text="Time since Kirdy Reset (Hr:Min:Sec)", tick_angle=-45, units="")
|
|
# Display the relative ts in custom %H:%M:%S format without local timezone
|
|
time_axis.tickStrings = tickStrings
|
|
# Prevent scaling prefix being added to the back fo axis label
|
|
time_axis.autoSIPrefix = False
|
|
time_axis.showLabel()
|
|
|
|
graph.setAxisItems({'bottom': time_axis})
|
|
graph.add_crosshair(pg.mkPen(color='red', width=1), {'color': 'green'})
|
|
|
|
#TODO: x_range should not be updated on every tick
|
|
graph.x_range_controller = LiveAxisRange(roll_on_tick=17, offset_left=4900)
|
|
graph.x_range_controller.crop_left_offset_to_data = True
|
|
# Enable linking of axes in the graph widget's context menu
|
|
graph.register(graph.getPlotItem().titleLabel.text) # Slight hack getting the title
|
|
|
|
self.max_samples = max_samples
|
|
ld_i_set_axis = LiveAxis('left', text="Current", units="A")
|
|
ld_i_set_axis.showLabel()
|
|
ld_i_set_graph.setAxisItems({'left': ld_i_set_axis})
|
|
ld_i_set_graph.addItem(self._ld_i_set_plot)
|
|
self.ld_i_set_connector = DataConnector(self._ld_i_set_plot, max_points=self.max_samples)
|
|
self.connectors += [self.ld_i_set_connector]
|
|
|
|
pd_mon_pwr_axis = LiveAxis('left', text="Power", units="W")
|
|
pd_mon_pwr_axis.showLabel()
|
|
pd_mon_pwr_graph.setAxisItems({'left': pd_mon_pwr_axis})
|
|
pd_mon_pwr_graph.addItem(self._pd_mon_pwr_plot)
|
|
self.pd_mon_pwr_connector = DataConnector(self._pd_mon_pwr_plot, max_points=self.max_samples)
|
|
self.connectors += [self.pd_mon_pwr_connector]
|
|
|
|
tec_temp_axis = LiveAxis('left', text="Temperature", units="°C")
|
|
tec_temp_axis.showLabel()
|
|
tec_temp_graph.setAxisItems({'left': tec_temp_axis})
|
|
tec_temp_graph.addItem(self._tec_setpoint_plot)
|
|
tec_temp_graph.addItem(self._tec_temp_plot)
|
|
self.tec_setpoint_connector = DataConnector(self._tec_setpoint_plot, max_points=1)
|
|
self.tec_temp_connector = DataConnector(self._tec_temp_plot, max_points=self.max_samples)
|
|
self.connectors += [self.tec_temp_connector, self.tec_setpoint_connector]
|
|
|
|
tec_i_axis = LiveAxis('left', text="Current", units="A")
|
|
tec_i_axis.showLabel()
|
|
tec_i_graph.setAxisItems({'left': tec_i_axis})
|
|
tec_i_graph.addLegend(brush=(50, 50, 200, 150))
|
|
tec_i_graph.y_range_controller = LiveAxisRange(fixed_range=[-1.0, 1.0])
|
|
tec_i_graph.addItem(self._tec_i_target_plot)
|
|
tec_i_graph.addItem(self._tec_i_measure_plot)
|
|
self.tec_i_target_connector = DataConnector(self._tec_i_target_plot, max_points=self.max_samples)
|
|
self.tec_i_measure_connector = DataConnector(self._tec_i_measure_plot, max_points=self.max_samples)
|
|
self.connectors += [self.tec_i_target_connector, self.tec_i_measure_connector]
|
|
|
|
def set_max_samples(self, max_samples):
|
|
self.max_samples = max_samples
|
|
for connector in self.connectors:
|
|
with connector.data_lock:
|
|
connector.max_points = self.max_samples
|
|
connector.x = deque(maxlen=int(connector.max_points))
|
|
connector.y = deque(maxlen=int(connector.max_points))
|
|
|
|
def plot_append(self, report):
|
|
try:
|
|
ld_i_set = report['laser']['ld_i_set']
|
|
pd_pwr = report['laser']['pd_pwr']
|
|
|
|
tec_i_set = report['thermostat']['i_set']
|
|
tec_i_measure = report['thermostat']['tec_i']
|
|
tec_temp = report['thermostat']['temperature']
|
|
|
|
ts = report['ts']
|
|
|
|
self.ld_i_set_connector.cb_append_data_point(ld_i_set, ts)
|
|
self.pd_mon_pwr_connector.cb_append_data_point(pd_pwr, ts)
|
|
|
|
if tec_temp is not None:
|
|
self.tec_temp_connector.cb_append_data_point(tec_temp, ts)
|
|
if self._temp_setpoint_line.isVisible():
|
|
self.tec_setpoint_connector.cb_append_data_point(self._temp_setpoint_line.value(), ts)
|
|
else:
|
|
self.tec_setpoint_connector.cb_append_data_point(tec_temp, ts)
|
|
if tec_i_measure is not None:
|
|
self.tec_i_measure_connector.cb_append_data_point(tec_i_measure, ts)
|
|
self.tec_i_target_connector.cb_append_data_point(tec_i_set, ts)
|
|
except Exception as e:
|
|
logging.error(f"Graph Value cannot be updated. Error:{e}. Data:{report}")
|
|
|
|
def clear_data_pts(self):
|
|
for connector in self.connectors:
|
|
connector.clear()
|
|
connector.resume()
|
|
|
|
def set_temp_setpoint_line(self, temp=None, visible=None):
|
|
if visible is not None:
|
|
self._temp_setpoint_line.setVisible(visible)
|
|
if temp is not None:
|
|
self._temp_setpoint_line.setValue(temp)
|
|
|
|
# PyQtGraph normally does not update this text when the line
|
|
# is not visible, so make sure that the temperature label
|
|
# gets updated always, and doesn't stay at an old value.
|
|
self._temp_setpoint_line.label.setText(f"{temp} °C", color='g')
|
|
|
|
class MutexParameter(pTypes.ListParameter):
|
|
"""
|
|
Mutually exclusive parameter where only one of its children is visible at a time, list selectable.
|
|
|
|
The ordering of the list items determines which children will be visible.
|
|
"""
|
|
def __init__(self, **opts):
|
|
super().__init__(**opts)
|
|
|
|
self.sigValueChanged.connect(self.show_chosen_child)
|
|
self.sigValueChanged.emit(self, self.opts['value'])
|
|
|
|
def _get_param_from_value(self, value):
|
|
if isinstance(self.opts['limits'], dict):
|
|
values_list = list(self.opts['limits'].values())
|
|
else:
|
|
values_list = self.opts['limits']
|
|
|
|
return self.children()[values_list.index(value)]
|
|
|
|
@pyqtSlot(object, object)
|
|
def show_chosen_child(self, value):
|
|
for param in self.children():
|
|
param.hide()
|
|
|
|
child_to_show = self._get_param_from_value(value.value())
|
|
child_to_show.show()
|
|
|
|
if child_to_show.opts.get('triggerOnShow', None):
|
|
child_to_show.sigValueChanged.emit(child_to_show, child_to_show.value())
|
|
|
|
registerParameterType('mutex', MutexParameter)
|
|
|
|
class NetSettingsForm(QtWidgets.QDialog, Ui_Conn_Settings_Form):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.setupUi(self)
|
|
|
|
def get_net_settings(self):
|
|
try:
|
|
ip_addr = [self.addr_in_0.text(), self.addr_in_1.text(), self.addr_in_2.text(), self.addr_in_3.text()]
|
|
for addr in ip_addr:
|
|
_ = int(addr)
|
|
|
|
return {
|
|
"ip_addr": ip_addr,
|
|
"port": int(self.port_in.text())
|
|
}
|
|
except ValueError:
|
|
return None
|
|
|
|
class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
|
|
"""The maximum number of sample points to store."""
|
|
DEFAULT_MAX_SAMPLES = 1000
|
|
DEFAULT_IP_ADDR = ['192', '168', '1', '128']
|
|
DEFAULT_PORT = 1337
|
|
|
|
LASER_DIODE_STATUS = [
|
|
{'name': 'Power', 'type': 'color', 'value': 'w', 'readonly': True},
|
|
{'name': 'Alarm', 'type': 'color', 'value': 'w', 'readonly': True},
|
|
]
|
|
|
|
LASER_DIODE_PARAMETERS = [
|
|
{'name': 'Readings', 'expanded': True, 'type': 'group', 'children': [
|
|
{'name': 'LD Current Set', 'type': 'float', 'suffix': 'A', 'siPrefix': True, 'readonly': True},
|
|
{'name': 'PD Current', 'type': 'float', 'suffix': 'A', 'siPrefix': True, 'readonly': True},
|
|
{'name': 'PD Power', 'type': 'float', 'suffix': 'W', 'siPrefix': True, 'readonly': True},
|
|
{'name': 'LF Mod Impedance', 'type': 'list', 'limits': ['Is50Ohm', 'Not50Ohm'], 'readonly': True}
|
|
]},
|
|
{'name': 'Output Config', 'expanded': True, 'type': 'group', 'children': [
|
|
{'name': 'LD Current Set', 'type': 'float', 'value': 0, 'step': 1, 'decimals': 6, 'limits': (0, 1),
|
|
'suffix': 'A', 'siPrefix': True, 'target': 'laser', 'action': 'set_i'},
|
|
{'name': 'LD Current Set Soft Limit', 'type': 'float', 'value': 0, 'step': 1, 'decimals': 6, 'limits': (0, 1),
|
|
'suffix': 'A', 'siPrefix': True, 'target': 'laser', 'action': 'set_i_soft_limit'},
|
|
{'name': 'LD Power Limit', 'type': 'float', 'value': 0, 'step': 1, 'decimals': 6, 'limits': (0, 0.3),
|
|
'suffix': 'W', 'siPrefix': True, 'target': 'laser', 'action': 'set_ld_pwr_limit'},
|
|
{'name': 'LD Terminals Short', 'type': 'bool', 'value': False, 'target': 'laser', 'action': 'set_ld_terms_short'},
|
|
{'name': 'Default Power On', 'type': 'bool', 'value': False, 'target': 'laser', 'action': 'set_default_pwr_on'},
|
|
]},
|
|
{'name': 'Photodiode Monitor Config', 'expanded': False, 'type': 'group', 'children': [
|
|
{'name': 'Responsitivity', 'type': 'float', 'value': 0, 'step': 1, 'decimals': 6, 'limits': (0, 3000),
|
|
'suffix': 'A/W', 'siPrefix': True, 'target': 'laser', 'action': 'set_pd_mon_responsitivity'},
|
|
{'name': 'Dark Current', 'type': 'float', 'value': 0, 'step': 1, 'decimals': 6, 'limits': (0, 3000),
|
|
'suffix': 'A', 'siPrefix': True, 'target': 'laser', 'action': 'set_pd_mon_dark_current'},
|
|
]},
|
|
]
|
|
|
|
THERMOSTAT_STATUS = [
|
|
{'name': 'Power', 'type': 'color', 'value': 'w', 'readonly': True},
|
|
{'name': 'Alarm', 'type': 'color', 'value': 'w', 'readonly': True},
|
|
]
|
|
|
|
THERMOSTAT_PARAMETERS = [
|
|
{'name': 'Readings', 'expanded': True, 'type': 'group', 'children': [
|
|
{'name': 'Temperature', 'type': 'float', 'format': '{value:.4f} °C', 'readonly': True},
|
|
{'name': 'Current through TEC', 'type': 'float', 'suffix': 'A', 'siPrefix': True, 'decimals': 6, 'readonly': True},
|
|
]},
|
|
{'name': 'Output Config', 'expanded': True, 'type': 'group', 'children': [
|
|
{'name': 'Control Method', 'type': 'mutex', 'limits': ['Constant Current', 'Temperature PID'],
|
|
'target_action_pair': [['thermostat', 'set_constant_current_control_mode'], ['thermostat', 'set_pid_control_mode']], 'children': [
|
|
{'name': 'Set Current', 'type': 'float', 'value': 0, 'step': 0.001, 'limits': (-1, 1), 'triggerOnShow': True,
|
|
'decimals': 6, 'suffix': 'A', 'siPrefix': True, 'target': 'thermostat', 'action': 'set_tec_i_out'},
|
|
{'name': 'Set Temperature', 'type': 'float', 'value': 25, 'step': 0.1, 'limits': (-273, 300),
|
|
'format': '{value:.4f} °C', 'target': 'thermostat', 'action': 'set_temperature_setpoint'},
|
|
]},
|
|
{'name': 'Limits', 'expanded': False, 'type': 'group', 'children': [
|
|
{'name': 'Max Cooling Current', 'type': 'float', 'value': 0, 'step': 0.001, 'decimals': 6, 'limits': (0, 1),
|
|
'suffix': 'A', 'siPrefix': True, 'target': 'thermostat', 'action': 'set_tec_max_cooling_i'},
|
|
{'name': 'Max Heating Current', 'type': 'float', 'value': 0, 'step': 0.001, 'decimals': 6, 'limits': (0, 1),
|
|
'suffix': 'A', 'siPrefix': True, 'target': 'thermostat', 'action': 'set_tec_max_heating_i'},
|
|
{'name': 'Max Voltage Difference', 'type': 'float', 'value': 0, 'step': 0.1, 'limits': (0, 5),
|
|
'suffix': 'V', 'siPrefix': True, 'target': 'thermostat', 'action': 'set_tec_max_v'},
|
|
]},
|
|
{'name': 'Default Power On', 'type': 'bool', 'value': False, 'target': 'thermostat', 'action': 'set_default_pwr_on'},
|
|
]},
|
|
# TODO Temperature ADC Filter Settings
|
|
{'name': 'Temperature Monitor Config', 'expanded': False, 'type': 'group', 'children': [
|
|
{'name': 'Upper Limit', 'type': 'float', 'value': 0, 'step': 1, 'decimals': 6, 'limits': (-273, 300),
|
|
'suffix': '°C', 'target': 'thermostat', 'action': 'set_temp_mon_upper_limit'},
|
|
{'name': 'Lower Limit', 'type': 'float', 'value': 0, 'step': 1, 'decimals': 6, 'limits': (-273, 300),
|
|
'suffix': '°C', 'target': 'thermostat', 'action': 'set_temp_mon_upper_limit'},
|
|
]},
|
|
{'name': 'Thermistor Settings','expanded': False, 'type': 'group', 'children': [
|
|
{'name': 'T₀', 'type': 'float', 'value': 0, 'step': 1, 'decimals': 6, 'limits': (-273, 300),
|
|
'suffix': '°C', 'target': 'thermostat', 'action': 'set_sh_t0'},
|
|
{'name': 'R₀', 'type': 'float', 'value': 0, 'step': 1, 'decimals': 6,
|
|
'suffix': 'Ω', 'siPrefix': True, 'target': 'thermostat', 'action': 'set_sh_r0'},
|
|
{'name': 'B', 'type': 'float', 'value': 3950, 'step': 1, 'suffix': 'K', 'decimals': 4, 'target': 'thermostat', 'action': 'set_sh_beta'},
|
|
]},
|
|
{'name': 'PID Config', 'expanded': False, 'type': 'group', 'children': [
|
|
{'name': 'Kp', 'type': 'float', 'step': 0.1, 'suffix': '', 'target': 'thermostat', 'action': 'set_pid_kp'},
|
|
{'name': 'Ki', 'type': 'float', 'step': 0.1, 'suffix': 'Hz', 'target': 'thermostat', 'action': 'set_pid_ki'},
|
|
{'name': 'Kd', 'type': 'float', 'step': 0.1, 'suffix': 's', 'target': 'thermostat', 'action': 'set_pid_kd'},
|
|
{'name': "PID Output Clamping", 'expanded': True, 'type': 'group', 'children': [
|
|
{'name': 'Minimum', 'type': 'float', 'step': 100, 'limits': (-1, 1), 'decimals': 6, 'suffix': 'A', 'target': 'thermostat', 'action': 'set_pid_output_min'},
|
|
{'name': 'Maximum', 'type': 'float', 'step': 100, 'limits': (-1, 1), 'decimals': 6, 'suffix': 'A', 'target': 'thermostat', 'action': 'set_pid_output_max'},
|
|
]},
|
|
# TODO PID AutoTune
|
|
]},
|
|
]
|
|
def __init__(self, args):
|
|
super().__init__()
|
|
self.kirdy = Kirdy()
|
|
self.setupUi(self)
|
|
|
|
self.ip_addr = self.DEFAULT_IP_ADDR
|
|
self.port = self.DEFAULT_PORT
|
|
|
|
self.net_settings_form = NetSettingsForm()
|
|
self.net_settings_form.accepted.connect(self.start_connecting)
|
|
|
|
self.max_samples = self.DEFAULT_MAX_SAMPLES
|
|
|
|
|
|
self._set_up_kirdy_menu()
|
|
self._set_up_ctrl_btns()
|
|
self._set_up_plot_menu()
|
|
|
|
self.params = [
|
|
Parameter.create(name=f"Laser Diode Status", type='group', value=0, children=self.LASER_DIODE_STATUS),
|
|
Parameter.create(name=f"Laser Diode Parameters", type='group', value=1, children=self.LASER_DIODE_PARAMETERS),
|
|
Parameter.create(name=f"Thermostat Status", type='group', value=2, children=self.THERMOSTAT_STATUS),
|
|
Parameter.create(name=f"Thermostat Parameters", type='group', value=3, children=self.THERMOSTAT_PARAMETERS),
|
|
]
|
|
self._set_param_tree()
|
|
|
|
self.tec_i_graph.setTitle("TEC Current")
|
|
self.tec_temp_graph.setTitle("TEC Temperature")
|
|
self.ld_i_set_graph.setTitle("LD Current Set")
|
|
self.pd_mon_pwr_graph.setTitle("PD Mon Power")
|
|
|
|
self.connect_btn.clicked.connect(self.show_net_settings_form)
|
|
self.report_box.stateChanged.connect(self.on_report_box_stateChanged)
|
|
|
|
self.kirdy_data_watcher = KirdyDataWatcher(self, self.kirdy, self.report_refresh_spin.value())
|
|
self.kirdy_data_watcher.connection_error_sig.connect(self.bail)
|
|
# TODO: Identify the usable range of set_update_s
|
|
self.report_apply_btn.clicked.connect(
|
|
lambda: self.kirdy_data_watcher.set_update_s(self.report_refresh_spin.value())
|
|
)
|
|
self.kirdy_data_watcher.setting_update_sig.connect(self.update_ld_ctrl_panel_settings)
|
|
self.kirdy_data_watcher.setting_update_sig.connect(self.update_thermostat_ctrl_panel_settings)
|
|
self.kirdy_data_watcher.report_update_sig.connect(self.update_ld_ctrl_panel_readings)
|
|
self.kirdy_data_watcher.report_update_sig.connect(self.update_thermostat_ctrl_panel_readings)
|
|
|
|
self.graphs = Graphs(self.ld_i_set_graph, self.pd_mon_pwr_graph, self.tec_i_graph, self.tec_temp_graph, max_samples=self.max_samples)
|
|
self.kirdy_data_watcher.report_update_sig.connect(self.graphs.plot_append)
|
|
|
|
self.save_flash_btn.clicked.connect(
|
|
lambda: self.save_ld_thermostat_settings_to_flash()
|
|
)
|
|
|
|
self.load_flash_btn.clicked.connect(
|
|
lambda: self.load_ld_thermostat_settings_from_flash()
|
|
)
|
|
|
|
self.loading_spinner.hide()
|
|
|
|
def show_net_settings_form(self):
|
|
self.net_settings_form.addr_in_0.setText(self.ip_addr[0])
|
|
self.net_settings_form.addr_in_1.setText(self.ip_addr[1])
|
|
self.net_settings_form.addr_in_2.setText(self.ip_addr[2])
|
|
self.net_settings_form.addr_in_3.setText(self.ip_addr[3])
|
|
self.net_settings_form.port_in.setText(str(self.port))
|
|
self.net_settings_form.show()
|
|
|
|
def _set_up_ctrl_btns(self):
|
|
@asyncSlot(bool)
|
|
async def ld_pwr_on(_):
|
|
await self.kirdy.laser.set_power_on(True)
|
|
self.ld_pwr_on_btn.clicked.connect(ld_pwr_on)
|
|
|
|
@asyncSlot(bool)
|
|
async def ld_pwr_off(_):
|
|
await self.kirdy.laser.set_power_on(False)
|
|
self.ld_pwr_off_btn.clicked.connect(ld_pwr_off)
|
|
|
|
@asyncSlot(bool)
|
|
async def ld_clear_alarm(_):
|
|
await self.kirdy.laser.clear_alarm()
|
|
self.ld_clear_alarm_btn.clicked.connect(ld_clear_alarm)
|
|
|
|
@asyncSlot(bool)
|
|
async def tec_pwr_on(_):
|
|
await self.kirdy.thermostat.set_power_on(True)
|
|
self.tec_pwr_on_btn.clicked.connect(tec_pwr_on)
|
|
|
|
@asyncSlot(bool)
|
|
async def tec_pwr_off(_):
|
|
await self.kirdy.thermostat.set_power_on(False)
|
|
self.tec_pwr_off_btn.clicked.connect(tec_pwr_off)
|
|
|
|
@asyncSlot(bool)
|
|
async def tec_clear_alarm(_):
|
|
await self.kirdy.thermostat.clear_alarm()
|
|
self.tec_clear_alarm_btn.clicked.connect(tec_clear_alarm)
|
|
|
|
# TODO: Create a Diag box with QtDesigner instead of setting up here
|
|
def _set_up_kirdy_menu(self):
|
|
self.kirdy_menu = QtWidgets.QMenu()
|
|
self.kirdy_menu.setTitle('Kirdy Settings')
|
|
|
|
@asyncSlot(int)
|
|
async def on_report_box_stateChanged(self, enabled):
|
|
await self.kirdy_data_watcher.set_report_mode(enabled)
|
|
|
|
@asyncSlot(bool)
|
|
async def reset_kirdy(_):
|
|
await self.kirdy.device.hard_reset()
|
|
await self._on_connection_changed(False)
|
|
await asyncio.sleep(0.1) # Wait for the reset to start
|
|
# TODO: Attempt to reconnect after resetting
|
|
|
|
self.actionReset.triggered.connect(reset_kirdy)
|
|
self.kirdy_menu.addAction(self.actionReset)
|
|
|
|
@asyncSlot(bool)
|
|
async def dfu_mode(_):
|
|
await self._on_connection_changed(False)
|
|
await self.kirdy.device.dfu()
|
|
|
|
self.actionEnter_DFU_Mode.triggered.connect(dfu_mode)
|
|
self.kirdy_menu.addAction(self.actionEnter_DFU_Mode)
|
|
|
|
# TODO: Add a form for user to set ip settings in multiple text boxes
|
|
@asyncSlot(bool)
|
|
async def network_settings(_):
|
|
ask_network = QtWidgets.QInputDialog(self)
|
|
ask_network.setWindowTitle("Network Settings")
|
|
ask_network.setLabelText("Set the kirdy's IPv4 address, port, prefix length and gateway")
|
|
ask_network.setTextValue("192.168.1.128 1337 24 192.168.1.1")
|
|
|
|
@pyqtSlot(str)
|
|
def set_ipv4(ipv4_settings):
|
|
sure = QtWidgets.QMessageBox(self)
|
|
sure.setWindowTitle("Set network?")
|
|
sure.setText(f"Setting this as network and disconnecting:<br>{ipv4_settings}")
|
|
|
|
@asyncSlot(object)
|
|
async def really_set(button):
|
|
addr, port, prefix_len, gateway = ipv4_settings.split()
|
|
addr = list(map(int, addr.split(".")))
|
|
gateway = list(map(int, gateway.split(".")))
|
|
|
|
await self.kirdy.device.set_ip_settings(addr, int(port), int(prefix_len), gateway)
|
|
|
|
# TODO: Add a dialogue box and ask if the user wanna reboot Kirdy immediately
|
|
await self.kirdy.device.hard_reset()
|
|
|
|
await self._on_connection_changed(False)
|
|
sure.buttonClicked.connect(really_set)
|
|
sure.show()
|
|
ask_network.textValueSelected.connect(set_ipv4)
|
|
ask_network.show()
|
|
|
|
self.actionNetwork_Settings.triggered.connect(network_settings)
|
|
self.kirdy_menu.addAction(self.actionNetwork_Settings)
|
|
|
|
@asyncSlot(bool)
|
|
async def load(_):
|
|
await self.kirdy.device.load_current_settings_to_flash()
|
|
loaded = QtWidgets.QMessageBox(self)
|
|
loaded.setWindowTitle("Config loaded")
|
|
loaded.setText(f"All channel configs have been loaded from flash.")
|
|
loaded.setIcon(QtWidgets.QMessageBox.Icon.Information)
|
|
loaded.show()
|
|
|
|
self.actionLoad_all_configs.triggered.connect(load)
|
|
self.kirdy_menu.addAction(self.actionLoad_all_configs)
|
|
|
|
@asyncSlot(bool)
|
|
async def save(_):
|
|
await self.kirdy.device.save_current_settings_to_flash()
|
|
saved = QtWidgets.QMessageBox(self)
|
|
saved.setWindowTitle("Config saved")
|
|
saved.setText(f"All channel configs have been saved to flash.")
|
|
saved.setIcon(QtWidgets.QMessageBox.Icon.Information)
|
|
saved.show()
|
|
self.actionSave_all_configs.triggered.connect(save)
|
|
self.kirdy_menu.addAction(self.actionSave_all_configs)
|
|
|
|
def about_kirdy():
|
|
# TODO: Replace the hardware revision placeholder
|
|
QtWidgets.QMessageBox.about(
|
|
self,
|
|
"About Kirdy",
|
|
f"""
|
|
<h1>Sinara 1550 Kirdy v"major rev"."minor rev"</h1>
|
|
"""
|
|
)
|
|
|
|
self.actionAbout_Kirdy.triggered.connect(about_kirdy)
|
|
self.kirdy_menu.addAction(self.actionAbout_Kirdy)
|
|
self.kirdy_settings.setMenu(self.kirdy_menu)
|
|
|
|
def _set_up_plot_menu(self):
|
|
self.plot_menu = QtWidgets.QMenu()
|
|
self.plot_menu.setTitle("Plot Settings")
|
|
|
|
clear = QtGui.QAction("Clear graphs", self.plot_menu)
|
|
clear.triggered.connect(self.clear_graphs)
|
|
self.plot_menu.addAction(clear)
|
|
self.plot_menu.clear = clear
|
|
|
|
self.samples_spinbox = QtWidgets.QSpinBox()
|
|
self.samples_spinbox.setRange(2, 100000)
|
|
self.samples_spinbox.setSuffix(' samples')
|
|
self.samples_spinbox.setValue(self.max_samples)
|
|
self.samples_spinbox.valueChanged.connect(self.set_max_samples)
|
|
|
|
limit_samples = QtWidgets.QWidgetAction(self.plot_menu)
|
|
limit_samples.setDefaultWidget(self.samples_spinbox)
|
|
self.plot_menu.addAction(limit_samples)
|
|
self.plot_menu.limit_samples = limit_samples
|
|
|
|
self.plot_settings.setMenu(self.plot_menu)
|
|
|
|
def _set_param_tree(self):
|
|
status = self.ld_status
|
|
status.setHeaderHidden(True)
|
|
status.setParameters(self.params[0], showTop=False)
|
|
|
|
tree = self.ld_tree
|
|
tree.setHeaderHidden(True)
|
|
tree.setParameters(self.params[1], showTop=False)
|
|
self.params[1].sigTreeStateChanged.connect(self.send_command)
|
|
|
|
status = self.tec_status
|
|
status.setHeaderHidden(True)
|
|
status.setParameters(self.params[2], showTop=False)
|
|
|
|
tree = self.tec_tree
|
|
tree.setHeaderHidden(True)
|
|
tree.setParameters(self.params[3], showTop=False)
|
|
self.params[3].sigTreeStateChanged.connect(self.send_command)
|
|
|
|
async def _on_connection_changed(self, result):
|
|
def ctrl_panel_setEnable(result):
|
|
self.ld_status.setEnabled(result)
|
|
self.ld_tree.setEnabled(result)
|
|
self.ld_pwr_on_btn.setEnabled(result)
|
|
self.ld_pwr_off_btn.setEnabled(result)
|
|
self.ld_clear_alarm_btn.setEnabled(result)
|
|
self.tec_status.setEnabled(result)
|
|
self.tec_tree.setEnabled(result)
|
|
self.tec_pwr_on_btn.setEnabled(result)
|
|
self.tec_pwr_off_btn.setEnabled(result)
|
|
self.tec_clear_alarm_btn.setEnabled(result)
|
|
ctrl_panel_setEnable(result)
|
|
|
|
def graph_group_setEnable(result):
|
|
self.ld_i_set_graph.setEnabled(result)
|
|
self.pd_mon_pwr_graph.setEnabled(result)
|
|
self.tec_i_graph.setEnabled(result)
|
|
self.tec_temp_graph.setEnabled(result)
|
|
graph_group_setEnable(result)
|
|
|
|
self.kirdy_settings.setEnabled(result)
|
|
self.report_refresh_spin.setEnabled(result)
|
|
|
|
self.report_group.setEnabled(result)
|
|
self.report_refresh_spin.setEnabled(result)
|
|
self.report_box.setEnabled(result)
|
|
self.report_apply_btn.setEnabled(result)
|
|
self.save_flash_btn.setEnabled(result)
|
|
self.load_flash_btn.setEnabled(result)
|
|
|
|
# TODO: Use QStateMachine to manage connections
|
|
self.connect_btn.clicked.disconnect()
|
|
if result:
|
|
self.connect_btn.setText("Disconnect")
|
|
self.connect_btn.clicked.connect(self.bail)
|
|
else:
|
|
self.connect_btn.setText("Connect")
|
|
self.connect_btn.clicked.connect(self.show_net_settings_form)
|
|
|
|
if result:
|
|
# TODO: self.hw_rev_data = await self.kirdy.hw_rev()
|
|
self._status()
|
|
self.kirdy_data_watcher.start_watching()
|
|
else:
|
|
pass
|
|
self.status_lbl.setText("Disconnected")
|
|
self.clear_graphs()
|
|
self.report_box.setChecked(False)
|
|
await self.kirdy_data_watcher.set_report_mode(False)
|
|
self.kirdy_data_watcher.stop_watching()
|
|
self.status_lbl.setText("Disconnected")
|
|
|
|
def _status(self):
|
|
# TODO: Get rev no from Kirdy and then add revision into the text
|
|
host = f"{self.ip_addr[0]}.{self.ip_addr[1]}.{self.ip_addr[2]}.{self.ip_addr[3]}"
|
|
port = self.port
|
|
self.status_lbl.setText(f"Connected to Kirdy @ {host}:{port}")
|
|
|
|
def clear_graphs(self):
|
|
self.graphs.clear_data_pts()
|
|
|
|
def update_network_settings(self):
|
|
net_settings = self.net_settings_form.get_net_settings()
|
|
if net_settings is None:
|
|
return False;
|
|
self.ip_addr = net_settings["ip_addr"]
|
|
self.port = net_settings["port"]
|
|
return True
|
|
|
|
|
|
@asyncSlot()
|
|
async def save_ld_thermostat_settings_to_flash(self):
|
|
await self.kirdy.device.save_current_settings_to_flash()
|
|
|
|
@asyncSlot()
|
|
async def load_ld_thermostat_settings_from_flash(self):
|
|
await self.kirdy.device.load_current_settings_from_flash()
|
|
|
|
@asyncSlot(dict)
|
|
async def graphs_update(self, report):
|
|
self.graphs.plot_append(report)
|
|
|
|
@asyncSlot(dict)
|
|
async def update_ld_ctrl_panel_settings(self, settings):
|
|
try:
|
|
settings = settings['laser']
|
|
with QSignalBlocker(self.params[1]):
|
|
self.params[1].child('Output Config', 'LD Current Set').setValue(settings["ld_drive_current"]['value'])
|
|
self.params[1].child('Output Config', 'LD Current Set Soft Limit').setValue(settings["ld_drive_current_limit"]['value'])
|
|
self.params[1].child('Output Config', 'LD Power Limit').setValue(settings["ld_pwr_limit"])
|
|
self.params[1].child('Output Config', 'LD Terminals Short').setValue(settings["ld_terms_short"])
|
|
self.params[1].child('Output Config', 'Default Power On').setValue(settings["default_pwr_on"])
|
|
if settings["pd_mon_params"]["responsitivity"] is not None:
|
|
self.params[1].child('Photodiode Monitor Config', 'Responsitivity').setValue(settings["pd_mon_params"]["responsitivity"])
|
|
else:
|
|
self.params[1].child('Photodiode Monitor Config', 'Responsitivity').setValue(0)
|
|
self.params[1].child('Photodiode Monitor Config', 'Dark Current').setValue(settings["pd_mon_params"]["i_dark"])
|
|
except Exception as e:
|
|
logging.error(f"Params tree cannot be updated. Error:{e}. Data:{settings}")
|
|
|
|
@asyncSlot(dict)
|
|
async def update_ld_ctrl_panel_readings(self, report):
|
|
try:
|
|
report = report['laser']
|
|
with QSignalBlocker(self.params[0]):
|
|
self.params[0].child('Power').setValue('g' if report['pwr_on'] else 'w')
|
|
self.params[0].child('Alarm').setValue('r' if report['pwr_excursion'] else 'w')
|
|
|
|
with QSignalBlocker(self.params[1]):
|
|
self.params[1].child('Readings', 'LD Current Set').setValue(report["ld_i_set"])
|
|
self.params[1].child('Readings', 'PD Current').setValue(report["pd_i"])
|
|
if report["pd_pwr"] is not None:
|
|
self.params[1].child('Readings', 'PD Power').setValue(report["pd_pwr"])
|
|
else:
|
|
self.params[1].child('Readings', 'PD Power').setValue(0)
|
|
self.params[1].child('Readings', 'LF Mod Impedance').setValue(report["term_status"])
|
|
except Exception as e:
|
|
logging.error(f"Params tree cannot be updated. Error:{e}. Data:{report}")
|
|
|
|
@asyncSlot(dict)
|
|
async def update_thermostat_ctrl_panel_settings(self, settings):
|
|
try:
|
|
settings = settings['thermostat']
|
|
with QSignalBlocker(self.params[3]):
|
|
self.params[3].child('Output Config', 'Control Method').setValue("Temperature PID" if settings["pid_engaged"] else "Constant Current")
|
|
self.params[3].child('Output Config', 'Control Method', 'Set Current').setValue(settings["tec_settings"]['i_set']['value'])
|
|
self.params[3].child('Output Config', 'Control Method', 'Set Temperature').setValue(float(settings["temperature_setpoint"]))
|
|
self.params[3].child('Output Config', 'Limits', 'Max Cooling Current').setValue(settings["tec_settings"]['max_i_pos']['value'])
|
|
self.params[3].child('Output Config', 'Limits', 'Max Heating Current').setValue(settings["tec_settings"]['max_i_neg']['value'])
|
|
self.params[3].child('Output Config', 'Limits', 'Max Voltage Difference').setValue(settings["tec_settings"]['max_v']['value'])
|
|
self.params[3].child('Output Config', 'Default Power On').setValue(settings["default_pwr_on"])
|
|
# TODO: Update the Temperature ADC Settings here as well
|
|
self.params[3].child('Temperature Monitor Config', 'Upper Limit').setValue(settings["temp_mon_settings"]['upper_limit'])
|
|
self.params[3].child('Temperature Monitor Config', 'Lower Limit').setValue(settings["temp_mon_settings"]['lower_limit'])
|
|
self.params[3].child('PID Config', 'Kp').setValue(settings["pid_params"]['kp'])
|
|
self.params[3].child('PID Config', 'Ki').setValue(settings["pid_params"]['ki'])
|
|
self.params[3].child('PID Config', 'Kd').setValue(settings["pid_params"]['kd'])
|
|
self.params[3].child('PID Config', 'PID Output Clamping', 'Minimum').setValue(settings["pid_params"]['output_min'])
|
|
self.params[3].child('PID Config', 'PID Output Clamping', 'Maximum').setValue(settings["pid_params"]['output_max'])
|
|
self.params[3].child('Thermistor Settings', 'T₀').setValue(settings["thermistor_params"]['t0'])
|
|
self.params[3].child('Thermistor Settings', 'R₀').setValue(settings["thermistor_params"]['r0'])
|
|
self.params[3].child('Thermistor Settings', 'B').setValue(settings["thermistor_params"]['b'])
|
|
self.graphs.set_temp_setpoint_line(temp=round(settings["temperature_setpoint"], 6))
|
|
self.graphs.set_temp_setpoint_line(visible=settings['pid_engaged'])
|
|
except Exception as e:
|
|
logging.error(f"Params tree cannot be updated. Error:{e}. Data:{settings}")
|
|
|
|
@asyncSlot(dict)
|
|
async def update_thermostat_ctrl_panel_readings(self, report):
|
|
try:
|
|
report = report['thermostat']
|
|
with QSignalBlocker(self.params[2]):
|
|
self.params[2].child('Power').setValue('g' if report['pwr_on'] else 'w')
|
|
self.params[2].child('Alarm').setValue('r' if report['temp_mon_status']['over_temp_alarm'] else 'w')
|
|
with QSignalBlocker(self.params[3]):
|
|
self.params[3].child('Readings', 'Temperature').setValue(report["temperature"])
|
|
self.params[3].child('Readings', 'Current through TEC').setValue(report["tec_i"])
|
|
except Exception as e:
|
|
logging.error(f"Params tree cannot be updated. Error:{e}. Data:{report}")
|
|
|
|
@pyqtSlot(int)
|
|
def set_max_samples(self, samples: int):
|
|
self.graphs.set_max_samples(samples)
|
|
|
|
@asyncSlot(int)
|
|
async def on_report_box_stateChanged(self, enabled):
|
|
await self.kirdy_data_watcher.set_report_mode(enabled)
|
|
|
|
@asyncSlot()
|
|
async def start_connecting(self):
|
|
if not self.update_network_settings():
|
|
self.status_lbl.setText("Invalid IP Settings Input")
|
|
return
|
|
host = f"{self.ip_addr[0]}.{self.ip_addr[1]}.{self.ip_addr[2]}.{self.ip_addr[3]}"
|
|
port = self.port
|
|
try:
|
|
if not (self.kirdy.connecting() or self.kirdy.connected()):
|
|
self.status_lbl.setText("Connecting...")
|
|
await self.kirdy.start_session(host=host, port=port, timeout=0.1)
|
|
await self._on_connection_changed(True)
|
|
else:
|
|
await self.bail()
|
|
except (OSError, TimeoutError) as e:
|
|
logging.error(f"Failed communicating to {host}:{port}: {e}")
|
|
await self.bail()
|
|
self.status_lbl.setText(f"Cannot connect to Kirdy@ {host}:{port}")
|
|
|
|
@asyncSlot()
|
|
async def bail(self):
|
|
await self._on_connection_changed(False)
|
|
await self.kirdy.end_session()
|
|
|
|
@asyncSlot(object, object)
|
|
async def send_command(self, param, changes):
|
|
for inner_param, change, data in changes:
|
|
if change == 'value':
|
|
""" cmd translation from mutex type parameter """
|
|
if inner_param.opts.get('target_action_pair', None) is not None:
|
|
target, action = inner_param.opts['target_action_pair'][inner_param.opts['limits'].index(data)]
|
|
cmd = getattr(getattr(self.kirdy, target), action)
|
|
await cmd()
|
|
continue
|
|
""" cmd translation from non-mutex type parameter"""
|
|
if inner_param.opts.get("target", None) is not None:
|
|
if inner_param.opts.get("action", None) is not None:
|
|
cmd = getattr(getattr(self.kirdy, inner_param.opts["target"]), inner_param.opts["action"])
|
|
await cmd(data)
|
|
continue
|
|
|
|
async def coro_main():
|
|
args = get_argparser().parse_args()
|
|
if args.logLevel:
|
|
logging.basicConfig(level=getattr(logging, args.logLevel))
|
|
|
|
app_quit_event = asyncio.Event()
|
|
|
|
app = QtWidgets.QApplication.instance()
|
|
app.aboutToQuit.connect(app_quit_event.set)
|
|
|
|
main_window = MainWindow(args)
|
|
main_window.show()
|
|
|
|
await app_quit_event.wait()
|
|
|
|
def main():
|
|
qasync.run(coro_main())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|