feat(univ_temp_cont): Introduction of Universal Temperature Controller
CI for debye_bec / test (push) Successful in 2m22s
CI for debye_bec / test (pull_request) Successful in 2m36s

This commit is contained in:
hitz_s committed 2026-10-22 15:21:54 +02:00
1 parent 0215d6ef6c
commit c142dedaaf
9 files changed
+753

No files matched your search

Whitespace-only changes.
@@ -0,0 +1,249 @@
"""
Universal Qt widgets
"""
from functools import partial
from bec_widgets.utils.colors import get_accent_colors
from qtpy.QtCore import Qt
# pylint: disable=E0611
from qtpy.QtGui import QFont
from qtpy.QtWidgets import (
QApplication,
QComboBox,
QDoubleSpinBox,
QGroupBox,
QHBoxLayout,
QLabel,
QPushButton,
QVBoxLayout,
QWidget,
)
LABEL_WIDTH = 118
ROW_MARGINS = (4, 0, 4, 0)
ROW_SPACING = 6
class Group(QGroupBox):
def __init__(self, label, widgets):
super().__init__(label)
self.layout = QVBoxLayout(self) # type: ignore
self.layout.setContentsMargins(6, 6, 6, 6)
self.layout.setSpacing(4)
for widget in widgets:
self.layout.addWidget(widget) # type: ignore
class NumberIndicator(QWidget):
def __init__(self, label="", unit=None, highlight=False, decimals=3):
super().__init__()
layout = QHBoxLayout(self)
layout.setContentsMargins(*ROW_MARGINS)
layout.setSpacing(ROW_SPACING)
self.label = QLabel(label)
self.label.setFixedWidth(LABEL_WIDTH)
self.label.setWordWrap(True)
layout.addWidget(self.label)
self.val = QLabel("-")
self.val.setAlignment(Qt.AlignTop) # type: ignore
layout.addWidget(self.val)
self.unit = unit
self.highlight = highlight
self.decimals = decimals
self.number = 0
if highlight:
font = QFont()
font.setBold(True)
font.setPointSize(14)
self.label.setFont(font)
self.val.setFont(font)
def value(self) -> float:
return self.number
def setLabel(self, label) -> None:
self.label.setText(label)
def setValue(self, number):
self.number = number
text = f"{number:.{int(self.decimals)}f}"
if self.unit is not None:
text = text + " " + self.unit
self.val.setText(text)
class InputNumberField(QWidget):
def __init__(
self,
identifier="",
label="",
unit=None,
prefix=None,
init=0.0,
decimals=1,
single_step=0.1,
ll=-1e6,
hl=1e6,
):
super().__init__()
layout = QHBoxLayout(self)
layout.setContentsMargins(*ROW_MARGINS)
layout.setSpacing(ROW_SPACING)
self.identifier = identifier
self.label = QLabel(label)
self.label.setFixedWidth(LABEL_WIDTH)
self.label.setWordWrap(True)
layout.addWidget(self.label)
self.val = QDoubleSpinBox()
self.val.setRange(ll, hl)
self.val.setDecimals(decimals)
self.val.setSingleStep(single_step)
self.val.setValue(init)
if unit is not None:
self.val.setSuffix(" " + unit)
if prefix is not None:
self.val.setPrefix(prefix + " ")
layout.addWidget(self.val)
self.ll = ll
self.hl = hl
def set_number(self, number):
self.val.setValue(number)
def set_range(self, ll, hl):
self.val.setRange(ll, hl)
self.ll = ll
self.hl = hl
def setUnit(self, unit: str):
self.val.setSuffix(" " + unit)
def get_range(self):
return (self.ll, self.hl)
def has_focus(self) -> bool:
return self.val.hasFocus()
def value(self) -> float:
return self.val.value()
def value_changed_connect(self, func):
"""Connect a function to the Enter/Return key press."""
self.val.valueChanged.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):
super().__init__()
layout = QHBoxLayout(self)
layout.setContentsMargins(*ROW_MARGINS)
layout.setSpacing(ROW_SPACING)
self.identifier = identifier
self.label = QLabel(label)
self.label.setFixedWidth(LABEL_WIDTH)
self.label.setWordWrap(True)
layout.addWidget(self.label)
self.value = QComboBox()
for entry in enums or []:
self.value.addItem(entry)
layout.addWidget(self.value)
def set_current_text(self, text):
self.value.setCurrentText(text)
def currentText(self) -> str:
return self.value.currentText()
def has_focus(self) -> bool:
return QApplication.focusWidget() is self.value.view()
def activated_connect(self, func):
"""Connect a function to the Enter/Return key press."""
self.value.activated.connect(
partial(
func,
identifier=self.identifier,
value_obj=self.value,
value=lambda: self.value.currentText(),
index=lambda: self.value.currentIndex(),
)
)
def setDisabled(self, disable):
self.value.setDisabled(disable)
def get_all_texts(self) -> list[str]:
return [self.value.itemText(i) for i in range(self.value.count())]
class Button(QWidget):
def __init__(
self, label=None, label_button: str = "", enabled: bool = False, color: str = "default"
):
super().__init__()
layout = QHBoxLayout(self)
layout.setContentsMargins(*ROW_MARGINS)
layout.setSpacing(ROW_SPACING)
if label is not None:
self.label = QLabel(label)
self.label.setFixedWidth(LABEL_WIDTH)
layout.addWidget(self.label)
self.button = QPushButton(label_button)
self.setEnabled(enabled)
layout.addWidget(self.button)
self.color = color
def clicked_connect(self, func):
"""Connect a function to the button press."""
self.button.clicked.connect(func)
def setEnabled(self, enable: bool = True):
self.button.setEnabled(enable)
self.apply_theme()
def setText(self, text):
self.button.setText(text)
def setColor(self, color: str):
self.color = color
self.apply_theme()
def apply_theme(self):
if self.button.isEnabled():
colors = get_accent_colors()
color = getattr(colors, self.color).name()
self.button.setStyleSheet(f"QPushButton {{ background-color: {color}; color: white; }}")
else:
self.button.setStyleSheet(
"QPushButton {{background-color: rgb(120, 120, 120); color: white;}}"
)
class TextIndicator(QWidget):
def __init__(self, label):
super().__init__()
layout = QHBoxLayout(self)
layout.setContentsMargins(*ROW_MARGINS)
layout.setSpacing(ROW_SPACING)
self.label = QLabel(label)
self.label.setFixedWidth(LABEL_WIDTH)
self.label.setWordWrap(True)
layout.addWidget(self.label)
self.text = QLabel("-")
self.text.setAlignment(Qt.AlignTop) # type: ignore
layout.addWidget(self.text)
def setLabel(self, label) -> None:
self.label.setText(label)
def setText(self, text):
self.text.setText(text)
def setColor(self, color: str):
self.text.setStyleSheet(f"QLabel {{color:{color}}}")
@@ -0,0 +1,15 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from .temperature_controller_plugin import TemperatureControllerPlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(TemperatureControllerPlugin())
if __name__ == "__main__": # pragma: no cover
main()
@@ -0,0 +1,248 @@
"""Temperature controller widget"""
from __future__ import annotations
import sys
import time
from collections import deque
import numpy as np
from bec_lib import bec_logger
from bec_lib.endpoints import MessageEndpoints
from bec_widgets.utils.bec_dispatcher import BECDispatcher
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme, get_accent_colors
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.widgets.plots.waveform.waveform import Waveform
# pylint: disable=E0611
from qtpy.QtCore import QTimer
from qtpy.QtWidgets import QApplication, QHBoxLayout, QVBoxLayout, QWidget
from ....devices.univ_temp_cont.univ_temp_cont_enums import (
RAMP_RATE_MAP,
UnivTempContHeaterTcFault,
UnivTempContOpMode,
UnivTempContPower,
UnivTempContRampRateUnit,
UnivTempContSensorBreak,
)
from .qt_widgets import ComboBox, Group, InputNumberField, NumberIndicator, TextIndicator
logger = bec_logger.logger
DEFAULT_WINDOW_S = 600
REFRESH_MS = 200
PLOT_SIGNALS = ["temperature", "setpoint", "output_power"]
READ_CONFIG_SIGNAL_MS = 1000
class TemperatureControllerError(Exception):
"""Temperature Controller specific error"""
class TemperatureController(BECWidget, QWidget):
"""Waveform with predefined curves and one status label."""
PLUGIN = True
ICON_NAME = "mode_heat"
def __init__(self, *arg, parent=None, **kwargs):
super().__init__(parent=parent, *arg, **kwargs)
self.get_bec_shortcuts()
if "univ_temp_cont" not in self.dev:
raise TemperatureControllerError(
"Universal Temperature Controller is not in the config, abort."
)
self._window_s = DEFAULT_WINDOW_S
root_layout = QHBoxLayout(self)
control_layout = QVBoxLayout()
control_layout.setContentsMargins(4, 4, 4, 4)
control_layout.setSpacing(4)
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.heater_temperature = NumberIndicator("Heater temperature", "°C", decimals=0)
self.heater_sensor_break = TextIndicator("Heater Thermocouple")
status_group = Group(
"Status",
[
self.sensor_break,
self.operating_mode,
self.output_power,
self.temperature,
self.heater_temperature,
self.heater_sensor_break,
],
)
self.power_mode = ComboBox("power_mode", "Power mode", UnivTempContPower)
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
)
self.ramp_rate_unit = ComboBox("ramp_rate_unit", "Ramp rate unit", UnivTempContRampRateUnit)
control_group = Group(
"Control",
[
self.power_mode,
self.target_setpoint,
self.setpoint,
self.ramp_rate,
self.ramp_rate_unit,
],
)
control_layout.addWidget(status_group)
control_layout.addWidget(control_group)
root_layout.addLayout(control_layout)
self.waveform = Waveform(parent=self, client=self.client, popups=False)
root_layout.addWidget(self.waveform, stretch=1)
self.waveform.set(title="Title", x_label="X-Label")
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
)
self.bec_dispatcher.connect_slot(
self._on_readback, MessageEndpoints.device_readback(self.dev.univ_temp_cont)
)
self._refresh_timer = QTimer(self)
self._refresh_timer.timeout.connect(self._redraw)
self._refresh_timer.start(REFRESH_MS)
self._config_signal_timer = QTimer(self)
self._config_signal_timer.timeout.connect(self._config_signal_read)
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.ramp_rate_unit.activated_connect(self._on_value_changed)
@SafeSlot()
def _on_value_changed(self, identifier, _, value, **kwargs):
logger.info(f"Got identifier {identifier} with value {value} and kwargs {kwargs}")
if "index" in kwargs.keys():
value = kwargs["index"]
self.dev.univ_temp_cont[identifier].put(value)
@SafeSlot(dict, dict)
def _on_readback(self, msg: dict, *_):
signals = msg.get("signals", {})
for signal, buf in self._buffers.items():
entry = signals.get(signal)
if entry is None or entry.get("value") is None:
continue
buf.append((entry["timestamp"], entry["value"]))
value = signals["temperature"]["value"]
self.temperature.setValue(value)
value = signals["heater_temperature"]["value"]
self.heater_temperature.setValue(value)
value = signals["output_power"]["value"]
self.output_power.setValue(value)
value = signals["sensor_break"]["value"]
match value:
case UnivTempContSensorBreak.GOOD:
self.sensor_break.setText("OK")
self.sensor_break.setColor(get_accent_colors().success.name())
case UnivTempContSensorBreak.SENSOR_BREAK:
self.sensor_break.setText("FAULT")
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"]
match value:
case UnivTempContHeaterTcFault.NO_FAULT:
self.heater_sensor_break.setText("OK")
self.heater_sensor_break.setColor(get_accent_colors().success.name())
case UnivTempContHeaterTcFault.TC_OPEN_FAULT:
self.heater_sensor_break.setText("OPEN")
self.heater_sensor_break.setColor(get_accent_colors().warning.name())
case (
UnivTempContHeaterTcFault.OVER_UNDERVOLTAGE_FAULT
| UnivTempContHeaterTcFault.TC_LOW_FAULT
| UnivTempContHeaterTcFault.TC_HIGH_FAULT
| UnivTempContHeaterTcFault.COLD_JUNC_LOW_FAULT
| UnivTempContHeaterTcFault.COLD_JUNC_HIGH_FAULT
| UnivTempContHeaterTcFault.TC_RANGE_FAULT
| UnivTempContHeaterTcFault.COLD_JUNC_RNG_FAULT
):
self.heater_sensor_break.setText("FAULT")
self.heater_sensor_break.setColor(get_accent_colors().emergency.name())
case _:
raise TemperatureControllerError(
f"Not implemented heater_sensor_break value {value}"
)
value = signals["op_mode"]["value"]
match value:
case UnivTempContOpMode.AUTO:
self.operating_mode.setText("AUTO")
self.operating_mode.setColor(get_accent_colors().success.name())
case UnivTempContOpMode.MANUAL:
self.operating_mode.setText("MANUAL")
self.operating_mode.setColor(get_accent_colors().warning.name())
case UnivTempContOpMode.STANDBY:
self.operating_mode.setText("STANDBY")
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"]
self.target_setpoint.set_number(value)
value = signals["setpoint"]["value"]
self.setpoint.setValue(value)
value = signals["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)
self.ramp_rate.setUnit(
list(RAMP_RATE_MAP.keys())[list(RAMP_RATE_MAP.values()).index(value)]
)
def _redraw(self):
now = time.time()
cutoff = now - self._window_s
for key, buf in self._buffers.items():
while buf and buf[0][0] < cutoff:
buf.popleft()
if not buf:
continue
t, v = zip(*buf)
self._curves[key].set_data(np.asarray(t) - now, np.asarray(v))
def cleanup(self):
self._refresh_timer.stop()
super().cleanup()
if __name__ == "__main__":
app = QApplication(sys.argv)
apply_theme("light")
dispatcher = BECDispatcher(gui_id="restart_server")
win = TemperatureController()
win.show()
sys.exit(app.exec_())
@@ -0,0 +1 @@
{'files': ['temperature_controller.py']}
@@ -0,0 +1,57 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from bec_widgets.utils.bec_designer import designer_material_icon
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from .temperature_controller import TemperatureController
DOM_XML = """
<ui language='c++'>
<widget class='TemperatureController' name='temperature_controller'>
</widget>
</ui>
"""
class TemperatureControllerPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
t = TemperatureController(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return ""
def icon(self):
return designer_material_icon(TemperatureController.ICON_NAME)
def includeFile(self):
return "temperature_controller"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "TemperatureController"
def toolTip(self):
return "TemperatureController"
def whatsThis(self):
return self.toolTip()
@@ -0,0 +1,14 @@
######################################
## Universal Temperature Controller ##
######################################
univ_temp_cont:
readoutPriority: monitored
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
softwareTrigger: false
@@ -0,0 +1,86 @@
"""Universal Temperature Controller"""
from __future__ import annotations
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,
)
if TYPE_CHECKING:
from bec_lib.devicemanager import ScanInfo
TIMEOUT_FOR_PV = 5
class UnivTempCont(PSIDeviceBase):
"""Class for the Universal Temperature Controller"""
temperature = Cpt(EpicsSignalRO, suffix="TEMPERATURE", kind="normal", 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(
EpicsSignal, suffix="TARGET-SETPOINT", kind="normal", doc="Target Setpoint"
)
output_power = Cpt(EpicsSignalRO, suffix="OUTPUT-POWER", kind="normal", doc="Output Power %")
ramp_rate = Cpt(EpicsSignal, suffix="RAMP-RATE", kind="normal", doc="Ramp rate")
pb = Cpt(EpicsSignal, suffix="PB", kind="config", doc="Controller PB value")
ti = Cpt(EpicsSignal, suffix="TI", kind="config", doc="Controller TI value")
td = Cpt(EpicsSignal, suffix="TD", kind="config", doc="Controller TD value")
op_mode = Cpt(EpicsSignal, suffix="OP-MODE", kind="normal", doc="Operation mode")
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")
autotune = Cpt(EpicsSignal, suffix="AUTOTUNE", kind="config", doc="Autotune function")
sensor_break = Cpt(
EpicsSignalRO, suffix="SENSOR-BREAK", kind="normal", doc="Thermocouple Status"
)
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")
connection = Cpt(
EpicsSignalRO, suffix="CONNECTION", kind="normal", doc="Connection to Eurotherm Controller"
)
heater_temperature = Cpt(
EpicsSignalRO,
suffix="HEATER-TEMPERATURE",
kind="normal",
doc="Temperature of heating element",
)
heater_tc_fault = Cpt(
EpicsSignalRO,
suffix="HEATER-TC-FAULT",
kind="normal",
doc="TC Fault inside heating element",
)
def __init__(self, *, name: str, prefix: str = "", scan_info: ScanInfo | None = None, **kwargs):
super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
self.timeout_for_pvwait = 5
# Wait for connection on all components, ensure IOC is connected
self.wait_for_connection(all_signals=True, timeout=5)
def on_connected(self) -> None:
"""
Called after the device is connected and its signals are connected.
Default values for signals should be set here.
"""
if self.setpoint_select != UnivTempContSetpointSelect.REMOTE:
self.setpoint_select.set(UnivTempContSetpointSelect.REMOTE).wait(
timeout=self.timeout_for_pvwait
)
@@ -0,0 +1,83 @@
"""Enums for the universal temperature controller"""
from enum import Enum
class UnivTempContOpMode(int, Enum):
"""Class for operating mode"""
AUTO = 0
MANUAL = 1
STANDBY = 2
class UnivTempContSetpointSelect(int, Enum):
"""Class for setpoint selection"""
LOCAL = 0
REMOTE = 1
class UnivTempContRampRateUnit(int, Enum):
"""Class for ramp rate unit selection"""
DEG_C_MIN = 0
DEG_C_H = 1
DEG_C_S = 2
RAMP_RATE_MAP = {
"°C/min": UnivTempContRampRateUnit.DEG_C_MIN,
"°C/h": UnivTempContRampRateUnit.DEG_C_H,
"°C/s": UnivTempContRampRateUnit.DEG_C_S,
}
class UnivTempContAutoTune(int, Enum):
"""Class for autotune function"""
NOTUNE = 0
TUNE = 1
class UnivTempContSensorBreak(int, Enum):
"Class for sensor break status"
GOOD = 0
SENSOR_BREAK = 1
class UnivTempContThermocouple(int, Enum):
"""Class for thermocouple type"""
TYPE_J = 0
Type_K = 1
class UnivTempContPower(int, Enum):
"""Class for power mode"""
AUTO = 0
STANDBY = 1
CONFIGURATION = 2
class UnivTempContConnection(int, Enum):
"""Class for connection status"""
DISCONNECTED = 0
CONNECTED = 1
class UnivTempContHeaterTcFault(int, Enum):
"""Class for the heater thermocouple fault"""
NO_FAULT = 0
TC_OPEN_FAULT = 1
OVER_UNDERVOLTAGE_FAULT = 2
TC_LOW_FAULT = 3
TC_HIGH_FAULT = 4
COLD_JUNC_LOW_FAULT = 5
COLD_JUNC_HIGH_FAULT = 6
TC_RANGE_FAULT = 7
COLD_JUNC_RNG_FAULT = 8