wip(univ_temp_cont)

This commit is contained in:
x01da committed 2026-10-26 12:52:11 +01:00
1 parent c142dedaaf
commit 37a17ed292
7 files changed
+115 -50

No files matched your search

+25
View File
@@ -19,6 +19,7 @@ _Widgets = {
"RestartServer": "RestartServer",
"ScanControlXAS": "ScanControlXAS",
"Scheduler": "Scheduler",
"TemperatureController": "TemperatureController",
}
@@ -242,3 +243,27 @@ class Scheduler(RPCBase):
"""
RPC-exposed: swap an item with the one directly after it.
"""
class TemperatureController(RPCBase):
"""Waveform with predefined curves and one status label."""
_IMPORT_MODULE = "debye_bec.bec_widgets.widgets.temperature_controller.temperature_controller"
@rpc_call
def remove(self):
"""
Cleanup the BECConnector
"""
@rpc_call
def attach(self):
"""
None
"""
@rpc_call
def detach(self):
"""
Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
"""
@@ -17,6 +17,10 @@ designer_plugins = {
"ScanControlXAS",
),
"Scheduler": ("debye_bec.bec_widgets.widgets.scheduler.scheduler", "Scheduler"),
"TemperatureController": (
"debye_bec.bec_widgets.widgets.temperature_controller.temperature_controller",
"TemperatureController",
),
}
widget_icons = {
@@ -26,4 +30,5 @@ widget_icons = {
"RestartServer": "restart_alt",
"ScanControlXAS": "tune",
"Scheduler": "assignment_add",
"TemperatureController": "mode_heat",
}
@@ -110,7 +110,8 @@ class InputNumberField(QWidget):
self.hl = hl
def set_number(self, number):
self.val.setValue(number)
if not self.val.hasFocus():
self.val.setValue(number)
def set_range(self, ll, hl):
self.val.setRange(ll, hl)
@@ -130,13 +131,21 @@ class InputNumberField(QWidget):
return self.val.value()
def value_changed_connect(self, func):
"""Connect a function to the Enter/Return key press."""
"""Connect a function to a change in value."""
self.val.valueChanged.connect(
partial(
func, identifier=self.identifier, value_obj=self.val, value=lambda: self.val.value()
)
)
def editing_finished_connect(self, func):
"""Connect a function to the Enter/Return press."""
self.val.editingFinished.connect(
partial(
func, identifier=self.identifier, value_obj=self.val, value=lambda: self.val.value()
)
)
class ComboBox(QWidget):
def __init__(self, identifier="", label="", enums=None):
@@ -155,7 +164,12 @@ class ComboBox(QWidget):
layout.addWidget(self.value)
def set_current_text(self, text):
self.value.setCurrentText(text)
if not QApplication.focusWidget() is self.value.view():
self.value.setCurrentText(text)
def setCurrentIndex(self, index):
if not QApplication.focusWidget() is self.value.view():
self.value.setCurrentIndex(index)
def currentText(self) -> str:
return self.value.currentText()
@@ -170,8 +184,8 @@ class ComboBox(QWidget):
func,
identifier=self.identifier,
value_obj=self.value,
value=lambda: self.value.currentText(),
index=lambda: self.value.currentIndex(),
value=lambda: self.value.currentText(),
)
)
@@ -7,6 +7,7 @@ import time
from collections import deque
import numpy as np
import pyqtgraph as pg
from bec_lib import bec_logger
from bec_lib.endpoints import MessageEndpoints
from bec_widgets.utils.bec_dispatcher import BECDispatcher
@@ -20,6 +21,7 @@ from qtpy.QtCore import QTimer
from qtpy.QtWidgets import QApplication, QHBoxLayout, QVBoxLayout, QWidget
from ....devices.univ_temp_cont.univ_temp_cont_enums import (
POWER_MAP,
RAMP_RATE_MAP,
UnivTempContHeaterTcFault,
UnivTempContOpMode,
@@ -31,6 +33,8 @@ from .qt_widgets import ComboBox, Group, InputNumberField, NumberIndicator, Text
logger = bec_logger.logger
DEVICE_NAME = "univ_temp_cont"
DEFAULT_WINDOW_S = 600
REFRESH_MS = 200
PLOT_SIGNALS = ["temperature", "setpoint", "output_power"]
@@ -51,7 +55,7 @@ class TemperatureController(BECWidget, QWidget):
super().__init__(parent=parent, *arg, **kwargs)
self.get_bec_shortcuts()
if "univ_temp_cont" not in self.dev:
if DEVICE_NAME not in self.dev:
raise TemperatureControllerError(
"Universal Temperature Controller is not in the config, abort."
)
@@ -67,7 +71,7 @@ class TemperatureController(BECWidget, QWidget):
self.sensor_break = TextIndicator("Thermocouple")
self.operating_mode = TextIndicator("Operating mode")
self.output_power = NumberIndicator("Output Power", "%", decimals=0)
self.temperature = NumberIndicator("Temperature", "°C", decimals=0)
self.temperature = NumberIndicator("Temperature", "°C", highlight=True, decimals=0)
self.heater_temperature = NumberIndicator("Heater temperature", "°C", decimals=0)
self.heater_sensor_break = TextIndicator("Heater Thermocouple")
@@ -83,15 +87,15 @@ class TemperatureController(BECWidget, QWidget):
],
)
self.power_mode = ComboBox("power_mode", "Power mode", UnivTempContPower)
self.power_mode = ComboBox("power", "Power mode", POWER_MAP.keys())
self.target_setpoint = InputNumberField(
"target_setpoint", "Target setpoint", "°C", decimals=0, single_step=1, ll=0, hl=850
)
self.setpoint = NumberIndicator("Setpoint", "°C", highlight=True, decimals=0)
self.ramp_rate = InputNumberField(
"ramp-rate", "Ramp rate", decimals=0, single_step=1, ll=0, hl=1e5
"ramp_rate", "Ramp rate", decimals=0, single_step=1, ll=0, hl=1e5
)
self.ramp_rate_unit = ComboBox("ramp_rate_unit", "Ramp rate unit", UnivTempContRampRateUnit)
self.ramp_rate_unit = ComboBox("ramp_rate_unit", "Ramp rate unit", RAMP_RATE_MAP.keys())
control_group = Group(
"Control",
@@ -106,24 +110,29 @@ class TemperatureController(BECWidget, QWidget):
control_layout.addWidget(status_group)
control_layout.addWidget(control_group)
control_layout.addStretch()
root_layout.addLayout(control_layout)
self.waveform = Waveform(parent=self, client=self.client, popups=False)
self.waveform = Waveform(parent=self, client=self.client, popups=True)
root_layout.addWidget(self.waveform, stretch=1)
self.waveform.set(title="Title", x_label="X-Label")
self.waveform.set(title="", x_label="Time")
self.waveform.enable_toolbar = False
self.waveform.enable_side_panel = False
self.waveform.plot_item.setAxisItems({"bottom": pg.DateAxisItem(orientation="bottom")})
self._buffers: dict[str, deque] = {}
self._curves = {}
for signal in PLOT_SIGNALS:
self._buffers[signal] = deque() # (timestamp, value), trimmed by time
self._curves[signal] = self.waveform.plot(
x=[], y=[], label=self.dev.univ_temp_cont.signal.full_name
x=[], y=[], label=getattr(self.dev.univ_temp_cont, signal).full_name
)
self._curves[signal].set_symbol(None)
self.bec_dispatcher.connect_slot(
self._on_readback, MessageEndpoints.device_readback(self.dev.univ_temp_cont)
self._on_readback, MessageEndpoints.device_readback(DEVICE_NAME)
)
self._refresh_timer = QTimer(self)
self._refresh_timer.timeout.connect(self._redraw)
@@ -134,32 +143,37 @@ class TemperatureController(BECWidget, QWidget):
self._config_signal_timer.start(READ_CONFIG_SIGNAL_MS)
self.power_mode.activated_connect(self._on_value_changed)
self.target_setpoint.value_changed_connect(self._on_value_changed)
self.ramp_rate.value_changed_connect(self._on_value_changed)
self.target_setpoint.editing_finished_connect(self._on_value_changed)
self.ramp_rate.editing_finished_connect(self._on_value_changed)
self.ramp_rate_unit.activated_connect(self._on_value_changed)
self.plot_init = True
@SafeSlot()
def _on_value_changed(self, identifier, _, value, **kwargs):
logger.info(f"Got identifier {identifier} with value {value} and kwargs {kwargs}")
def _on_value_changed(self, *_, **kwargs):
if "index" in kwargs.keys():
value = kwargs["index"]
self.dev.univ_temp_cont[identifier].put(value)
value = kwargs["index"]()
else:
value = kwargs["value"]()
signal = getattr(self.dev.univ_temp_cont, kwargs["identifier"])
signal.put(value)
@SafeSlot(dict, dict)
def _on_readback(self, msg: dict, *_):
signals = msg.get("signals", {})
now = time.time()
for signal, buf in self._buffers.items():
entry = signals.get(signal)
entry = signals.get(DEVICE_NAME + "_" + signal)
if entry is None or entry.get("value") is None:
continue
buf.append((entry["timestamp"], entry["value"]))
value = signals["temperature"]["value"]
buf.append((now, entry["value"]))
value = signals[DEVICE_NAME + "_" + "temperature"]["value"]
self.temperature.setValue(value)
value = signals["heater_temperature"]["value"]
value = signals[DEVICE_NAME + "_" + "heater_temperature"]["value"]
self.heater_temperature.setValue(value)
value = signals["output_power"]["value"]
value = signals[DEVICE_NAME + "_" + "output_power"]["value"]
self.output_power.setValue(value)
value = signals["sensor_break"]["value"]
value = signals[DEVICE_NAME + "_" + "sensor_break"]["value"]
match value:
case UnivTempContSensorBreak.GOOD:
self.sensor_break.setText("OK")
@@ -169,7 +183,7 @@ class TemperatureController(BECWidget, QWidget):
self.sensor_break.setColor(get_accent_colors().emergency.name())
case _:
raise TemperatureControllerError(f"Not implemented sensor_break value {value}")
value = signals["heater_tc_fault"]["value"]
value = signals[DEVICE_NAME + "_" + "heater_tc_fault"]["value"]
match value:
case UnivTempContHeaterTcFault.NO_FAULT:
self.heater_sensor_break.setText("OK")
@@ -192,7 +206,7 @@ class TemperatureController(BECWidget, QWidget):
raise TemperatureControllerError(
f"Not implemented heater_sensor_break value {value}"
)
value = signals["op_mode"]["value"]
value = signals[DEVICE_NAME + "_" + "op_mode"]["value"]
match value:
case UnivTempContOpMode.AUTO:
self.operating_mode.setText("AUTO")
@@ -205,22 +219,23 @@ class TemperatureController(BECWidget, QWidget):
self.operating_mode.setColor(get_accent_colors().default.name())
case _:
raise TemperatureControllerError(f"Not implemented op_mode value {value}")
value = signals["power"]["value"]
self.power_mode.set_current_text(value)
value = signals["target_setpoint"]["value"]
index = signals[DEVICE_NAME + "_" + "power"]["value"]
self.power_mode.setCurrentIndex(index)
value = signals[DEVICE_NAME + "_" + "target_setpoint"]["value"]
self.target_setpoint.set_number(value)
value = signals["setpoint"]["value"]
value = signals[DEVICE_NAME + "_" + "setpoint"]["value"]
self.setpoint.setValue(value)
value = signals["ramp_rate"]["value"]
value = signals[DEVICE_NAME + "_" + "ramp_rate"]["value"]
self.ramp_rate.set_number(value)
def _config_signal_read(self):
value = self.dev.univ_temp_cont.ramp_rate_unit.read()
if value is None:
raise TemperatureControllerError("Reading ramp rate unit returned None!")
self.ramp_rate_unit.set_current_text(value)
index = value[DEVICE_NAME + "_ramp_rate_unit"]["value"]
self.ramp_rate_unit.setCurrentIndex(index)
self.ramp_rate.setUnit(
list(RAMP_RATE_MAP.keys())[list(RAMP_RATE_MAP.values()).index(value)]
list(RAMP_RATE_MAP.keys())[list(RAMP_RATE_MAP.values()).index(index)]
)
def _redraw(self):
@@ -232,7 +247,10 @@ class TemperatureController(BECWidget, QWidget):
if not buf:
continue
t, v = zip(*buf)
self._curves[key].set_data(np.asarray(t) - now, np.asarray(v))
self._curves[key].set_data(np.asarray(t), np.asarray(v))
if len(self._buffers[PLOT_SIGNALS[0]]) > 0 and self.plot_init:
self.plot_init = False
self.waveform.plot_item.enableAutoRange(axis=pg.ViewBox.XYAxes, enable=True)
def cleanup(self):
self._refresh_timer.stop()
@@ -7,7 +7,6 @@ univ_temp_cont:
description: Universal Temperature Controller
deviceClass: debye_bec.devices.univ_temp_cont.univ_temp_cont.UnivTempCont
deviceConfig:
auto_monitor: true
prefix: "X01DA-ES-CRS:"
onFailure: retry
enabled: true
@@ -7,17 +7,8 @@ from typing import TYPE_CHECKING
from ophyd import Component as Cpt
from ophyd import EpicsSignal, EpicsSignalRO
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from univ_temp_cont_enums import (
UnivTempContAutoTune,
UnivTempContConnection,
UnivTempContHeaterTcFault,
UnivTempContOpMode,
UnivTempContPower,
UnivTempContRampRateUnit,
UnivTempContSensorBreak,
UnivTempContSetpointSelect,
UnivTempContThermocouple,
)
from .univ_temp_cont_enums import UnivTempContSetpointSelect
if TYPE_CHECKING:
from bec_lib.devicemanager import ScanInfo
@@ -28,7 +19,9 @@ TIMEOUT_FOR_PV = 5
class UnivTempCont(PSIDeviceBase):
"""Class for the Universal Temperature Controller"""
temperature = Cpt(EpicsSignalRO, suffix="TEMPERATURE", kind="normal", doc="Temperature")
temperature = Cpt(
EpicsSignalRO, suffix="TEMPERATURE", kind="normal", auto_monitor=True, doc="Temperature"
)
timestamp = Cpt(EpicsSignalRO, suffix="TIMESTAMP", kind="omitted", doc="Current timestamp")
setpoint = Cpt(EpicsSignalRO, suffix="SETPOINT", kind="normal", doc="Setpoint on Controller")
target_setpoint = Cpt(
@@ -43,7 +36,9 @@ class UnivTempCont(PSIDeviceBase):
setpoint_select = Cpt(
EpicsSignal, suffix="SETPOINT-SELECT", kind="config", doc="Local or remote setpoint"
)
ramp_rate_unit = Cpt(EpicsSignal, suffix="RAMP-RATE-UNIT", kind="config", doc="Ramp rate unit")
ramp_rate_unit = Cpt(
EpicsSignal, suffix="RAMP-RATE-UNIT", kind="config", auto_monitor=True, doc="Ramp rate unit"
)
autotune = Cpt(EpicsSignal, suffix="AUTOTUNE", kind="config", doc="Autotune function")
sensor_break = Cpt(
EpicsSignalRO, suffix="SENSOR-BREAK", kind="normal", doc="Thermocouple Status"
@@ -51,7 +46,9 @@ class UnivTempCont(PSIDeviceBase):
thermocouple = Cpt(EpicsSignal, suffix="THERMOCOUPLE", kind="config", doc="Thermocouple type")
power = Cpt(EpicsSignal, suffix="POWER", kind="normal", doc="Power status")
controller_type = Cpt(EpicsSignalRO, suffix="TYPE", kind="config", doc="Controller type")
heartbeat = Cpt(EpicsSignalRO, suffix="HEARTBEAT", kind="normal", doc="Heartbeat")
heartbeat = Cpt(
EpicsSignalRO, suffix="HEARTBEAT", kind="normal", auto_monitor=True, doc="Heartbeat"
)
connection = Cpt(
EpicsSignalRO, suffix="CONNECTION", kind="normal", doc="Connection to Eurotherm Controller"
)
@@ -62,6 +62,13 @@ class UnivTempContPower(int, Enum):
CONFIGURATION = 2
POWER_MAP = {
"Auto": UnivTempContPower.AUTO,
"Standby": UnivTempContPower.STANDBY,
"Configuration": UnivTempContPower.CONFIGURATION,
}
class UnivTempContConnection(int, Enum):
"""Class for connection status"""