wip(temperature_controller, scheduler)

debye_bec/bec_widgets/widgets/scheduler/item_dialog.py

@@ -73,12 +73,14 @@ class ScheduleItemDialog(QDialog):
             from ..digital_twin.digital_twin import DigitalTwin
             from ..edge_selector import EdgeSelector
             from ..scan_control_xas.scan_control_xas import ScanControlXAS
+            from ..temperature_controller.temperature_controller import TemperatureControllerControl
 
             self.DigitalTwinCore = DigitalTwinCore
             self.DigitalTwin = DigitalTwin
             self.AutoGain = AutoGain
             self.EdgeSelector = EdgeSelector
             self.ScanControlXAS = ScanControlXAS
+            self.TemperatureController = TemperatureControllerControl
 
         layout = QVBoxLayout(self)
 
@@ -96,6 +98,7 @@ class ScheduleItemDialog(QDialog):
             self._build_ic_page()
             self._build_reffoil_page()
             self._build_auto_gain_page()
+            self._build_temperature_controller_page()
         self._build_other_page()
 
         self.item_combo.currentIndexChanged.connect(self.stack.setCurrentIndex)
@@ -118,6 +121,7 @@ class ScheduleItemDialog(QDialog):
             "ic": self._collect_ic_result,
             "reffoil": self._collect_reffoil_result,
             "auto_gain": self._collect_auto_gain_result,
+            "temperautre_control": self._collect_temperature_controller_result,
             "other": self._collect_custom_result,
         }
 
@@ -535,6 +539,69 @@ class ScheduleItemDialog(QDialog):
     def _prefill_auto_gain_form(self, state: dict):
         self.auto_gain.load_config(state["config"])
 
+    # ------------------------------------------------------------------ #
+    # Temperature Controller
+    # ------------------------------------------------------------------ #
+    def _build_tempature_controller_page(self):
+        page = QWidget()
+        outer = QVBoxLayout(page)
+
+        scroll = QScrollArea()
+        scroll.setWidgetResizable(True)
+        self.temperature_controller = self.TemperatureController(config_only=True, parent=scroll, client=self._client)
+
+        scroll.setWidget(self.temperature_controller)
+        outer.addWidget(scroll)
+
+        # hint = QLabel(
+        #     "Configure the auto-gain functionality, then confirm with Add below - the "
+        #     "auto-gain will only be executed when this schedule item runs, not immediately."
+        # )
+        # hint.setWordWrap(True)
+        # hint.setStyleSheet("color: gray;")
+        # outer.addWidget(hint)
+
+        self._add_page("temperature_controller", "Temperature Controller", page)
+
+    def _collect_temperature_controller_result(self) -> dict:
+        config = self.temperature_controller.get_config()
+
+        cmd = f"dev.univ_temp_cont.load_config(config={config})"
+        return {
+            "command": cmd,
+            "kind": "custom",
+            "form_state": {"text": cmd, "source": "temperature_controller", "config": config},
+        }
+
+    def _prefill_temperature_controller_form(self, state: dict):
+        self.temperature_controller.load_config(state["config"])
+
+    # ------------------------------------------------------------------ #
+    # Sleep
+    # ------------------------------------------------------------------ #
+    def _build_sleep_page(self):
+        page = QWidget()
+        form = QFormLayout(page)
+        form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.FieldsStayAtSizeHint)
+
+        self.sleep_spin = QDoubleSpinBox(suffix=" s")
+        form.addRow("Sleep for", self.sleep_spin)
+
+        self._add_page("sleep", "Sleep", page)
+
+    def _collect_sleep_result(self) -> dict:
+        value = self.sleep_spin.value()
+        return {
+            "command": f"time.sleep({value})",
+            "kind": "sleep",
+            "form_state": {
+                "value": value,
+            },
+        }
+
+    def _prefill_sleep_page(self, state: dict):
+        self.sleep_spin.setValue(float(state.get("value", 0.0)))
+
     # ------------------------------------------------------------------ #
     # Other
     # ------------------------------------------------------------------ #
@@ -591,6 +658,7 @@ class ScheduleItemDialog(QDialog):
             self._prefill_reffoil_form(state)
         elif kind == "custom" and source == "auto_gain" and self._select_page("auto_gain"):
             self._prefill_auto_gain_form(state)
+        elif kind == "custom" and source == "temperature_controller" and self._select_page("temperature_controller")
         elif initial.get("command"):
             # "custom" kind with no (recognized/available) source, or a
             # legacy/unrecognized item - fall back to showing the raw

debye_bec/bec_widgets/widgets/scheduler/scheduler.py

@@ -143,6 +143,18 @@ def build_item_display_text(item: ScheduleItem, beamline: str | None) -> str:
                     + "}"
                 )
                 text += "\nProhibited gains: " + prohibited_gains
+        elif item.command.startswith("dev.univ_temp_cont.load_config("):
+            config = item.form_state["config"]
+            match config["power_mode"]:
+                case "Standby":
+                    text += "Switch temperature controller to standby"
+                case "Configuration":
+                    text += "Switch temperature controller to configuration mode"
+                case _:
+                    text += (
+                        f"Set temperature controller target setpoint to {config['target_setpoint']}"
+                        + f" °C\nwith ramp rate of {config['rampe_rate']} {config['ramp_rate_unit']}"
+                    )
         elif item.command.startswith("dev.abs."):
             action = item.form_state["action"]
             text += f"{action} Absorber"

debye_bec/bec_widgets/widgets/temperature_controller/qt_widgets.py

@@ -171,6 +171,9 @@ class ComboBox(QWidget):
         if not QApplication.focusWidget() is self.value.view():
             self.value.setCurrentIndex(index)
 
+    def currentIndex(self):
+        return self.value.currentIndex()
+
     def currentText(self) -> str:
         return self.value.currentText()
 

debye_bec/bec_widgets/widgets/temperature_controller/temperature_controller.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import sys
 import time
 from collections import deque
+from typing import Literal
 
 import numpy as np
 import pyqtgraph as pg
@@ -15,11 +16,16 @@ from bec_widgets.utils.bec_widget import BECWidget
 from bec_widgets.utils.colors import apply_theme
 from bec_widgets.utils.error_popups import SafeSlot
 from bec_widgets.widgets.plots.waveform.waveform import Waveform
+from pydantic import BaseModel
+from PySide6.QtCore import QLocale, QPoint, QRect, QSize, Qt
+from PySide6.QtGui import QCursor, QFont, QIcon, QPalette, QRegion
+from PySide6.QtWidgets import QSizePolicy
 
 # pylint: disable=E0611
 from qtpy.QtCore import QTimer
 from qtpy.QtWidgets import QApplication, QHBoxLayout, QLabel, QVBoxLayout, QWidget
 
+from ....devices.univ_temp_cont.univ_temp_cont import ConfigDict
 from ....devices.univ_temp_cont.univ_temp_cont_enums import (
     POWER_MAP,
     RAMP_RATE_MAP,
@@ -48,6 +54,61 @@ class TemperatureControllerError(Exception):
     """Temperature Controller specific error"""
 
 
+class TemperatureControllerControl(QWidget):
+    def __init__(self, config_only: bool = False, *arg, parent=None, **kwargs) -> None:
+        super().__init__(*arg, parent=parent, **kwargs)
+
+        self._layout = QVBoxLayout(self)
+
+        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.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", RAMP_RATE_MAP.keys())
+
+        if not config_only:
+            self.setpoint = NumberIndicator("Setpoint", "°C", highlight=True, decimals=0)
+            widgets = [
+                self.power_mode,
+                self.target_setpoint,
+                self.setpoint,
+                self.ramp_rate,
+                self.ramp_rate_unit,
+            ]
+        else:
+            widgets = [self.power_mode, self.target_setpoint, self.ramp_rate, self.ramp_rate_unit]
+
+        control_group = Group("Control", widgets)
+        self.ramp_rate_unit.activated_connect(self._on_ramp_rate_unit_change)
+        self._on_ramp_rate_unit_change()
+
+        self._layout.addWidget(control_group)
+
+    @SafeSlot()
+    def _on_ramp_rate_unit_change(self, *_, **__):
+        index = self.ramp_rate_unit.currentIndex()
+        self.ramp_rate.setUnit(
+            list(RAMP_RATE_MAP.keys())[list(RAMP_RATE_MAP.values()).index(index)]
+        )
+
+    def get_config(self) -> ConfigDict:
+        return ConfigDict(
+            power_mode=self.power_mode.currentText(),
+            target_setpoint=self.target_setpoint.value(),
+            ramp_rate=self.ramp_rate.value(),
+            ramp_rate_unit=self.ramp_rate_unit.currentText(),
+        )
+
+    def load_config(self, config: ConfigDict) -> None:
+        self.power_mode.setCurrentIndex(POWER_MAP[config.power_mode])
+        self.ramp_rate_unit.setCurrentIndex(RAMP_RATE_MAP[config.ramp_rate_unit])
+        self.ramp_rate.set_number(config.ramp_rate)
+        self.target_setpoint.set_number(config.target_setpoint)
+
+
 class TemperatureController(BECWidget, QWidget):
     """Waveform with predefined curves and one status label."""
 
@@ -65,6 +126,42 @@ class TemperatureController(BECWidget, QWidget):
 
         self._window_s = DEFAULT_WINDOW_S
 
+        self.init_ui()
+
+        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=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(DEVICE_NAME)
+        )
+        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.control.power_mode.activated_connect(self._on_value_changed)
+        self.control.target_setpoint.editing_finished_connect(self._on_value_changed)
+        self.control.ramp_rate.editing_finished_connect(self._on_value_changed)
+        self.control.ramp_rate_unit.activated_connect(self._on_value_changed)
+        self.pid_pb.editing_finished_connect(self._on_value_changed)
+        self.pid_ti.editing_finished_connect(self._on_value_changed)
+        self.pid_td.editing_finished_connect(self._on_value_changed)
+        self.autotune.clicked_connect(self._start_stop_autotune)
+        self.thermocouple.activated_connect(self._on_value_changed)
+
+        self.plot_init = True
+
+    def init_ui(self) -> None:
+        """Init the UI for the temperature controller widget, should only take place once."""
         root_layout = QHBoxLayout(self)
 
         control_layout = QVBoxLayout()
@@ -90,26 +187,7 @@ class TemperatureController(BECWidget, QWidget):
             ],
         )
 
-        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
-        )
-        self.ramp_rate_unit = ComboBox("ramp_rate_unit", "Ramp rate unit", RAMP_RATE_MAP.keys())
-
-        control_group = Group(
-            "Control",
-            [
-                self.power_mode,
-                self.target_setpoint,
-                self.setpoint,
-                self.ramp_rate,
-                self.ramp_rate_unit,
-            ],
-        )
+        self.control = TemperatureControllerControl()
 
         self.pid_pb = InputNumberField("pb", "PB", decimals=0, single_step=1, ll=1, hl=9999)
         self.pid_ti = InputNumberField(
@@ -132,7 +210,7 @@ class TemperatureController(BECWidget, QWidget):
         configuration_group = Group("Configuration", [self.tc_info, self.thermocouple])
 
         control_layout.addWidget(status_group)
-        control_layout.addWidget(control_group)
+        control_layout.addWidget(self.control)
         control_layout.addWidget(pid_loop_group)
         control_layout.addWidget(configuration_group)
         control_layout.addStretch()
@@ -141,10 +219,6 @@ class TemperatureController(BECWidget, QWidget):
 
         waveform_layout = QVBoxLayout()
 
-        self.waveform = Waveform(parent=self, client=self.client, popups=True)
-
-        waveform_layout.addWidget(self.waveform, stretch=1)
-
         waveform_settings_layout = QHBoxLayout()
         self.user_window_s = InputNumberField(
             "",
@@ -161,6 +235,10 @@ class TemperatureController(BECWidget, QWidget):
 
         waveform_layout.addLayout(waveform_settings_layout)
 
+        self.waveform = Waveform(parent=self, client=self.client, popups=True)
+
+        waveform_layout.addWidget(self.waveform, stretch=1)
+
         root_layout.addLayout(waveform_layout, stretch=1)
 
         self.waveform.set(title="", x_label="Time")
@@ -168,38 +246,6 @@ class TemperatureController(BECWidget, QWidget):
         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=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(DEVICE_NAME)
-        )
-        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.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.pid_pb.editing_finished_connect(self._on_value_changed)
-        self.pid_ti.editing_finished_connect(self._on_value_changed)
-        self.pid_td.editing_finished_connect(self._on_value_changed)
-        self.autotune.clicked_connect(self._start_stop_autotune)
-        self.thermocouple.activated_connect(self._on_value_changed)
-
-        self.plot_init = True
-
     @SafeSlot()
     def apply_theme(self, theme):
         self.sensor_break.apply_theme()
@@ -209,6 +255,20 @@ class TemperatureController(BECWidget, QWidget):
         self.autotune.apply_theme()
         return super().apply_theme(theme)
 
+    def load_config(self, config: ConfigDict) -> None:
+        # Power mode
+        signal = getattr(self.dev.univ_temp_cont, "power")
+        signal.put(POWER_MAP[config.power_mode])
+        # Ramp rate unit
+        signal = getattr(self.dev.univ_temp_cont, "ramp_rate_unit")
+        signal.put(RAMP_RATE_MAP[config.ramp_rate_unit])
+        # Ramp rate
+        signal = getattr(self.dev.univ_temp_cont, "ramp_rate")
+        signal.put(config.ramp_rate)
+        # Target setpoint
+        signal = getattr(self.dev.univ_temp_cont, "target_setpoint")
+        signal.put(config.target_setpoint)
+
     @SafeSlot()
     def _on_value_changed(self, *_, **kwargs):
         if "index" in kwargs.keys():
@@ -286,22 +346,19 @@ class TemperatureController(BECWidget, QWidget):
             case _:
                 raise TemperatureControllerError(f"Not implemented op_mode value {value}")
         index = signals[DEVICE_NAME + "_" + "power"]["value"]
-        self.power_mode.setCurrentIndex(index)
+        self.control.power_mode.setCurrentIndex(index)
         value = signals[DEVICE_NAME + "_" + "target_setpoint"]["value"]
-        self.target_setpoint.set_number(value)
+        self.control.target_setpoint.set_number(value)
         value = signals[DEVICE_NAME + "_" + "setpoint"]["value"]
-        self.setpoint.setValue(value)
+        self.control.setpoint.setValue(value)
         value = signals[DEVICE_NAME + "_" + "ramp_rate"]["value"]
-        self.ramp_rate.set_number(value)
+        self.control.ramp_rate.set_number(value)
 
     def _config_signal_read(self):
         # Ramp rate unit
         value = self.dev.univ_temp_cont.ramp_rate_unit.read()
         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(index)]
-        )
+        self.control.ramp_rate_unit.setCurrentIndex(index)
         # PB
         value = self.dev.univ_temp_cont.pb.read()[DEVICE_NAME + "_pb"]["value"]
         self.pid_pb.set_number(value)
@@ -359,6 +416,6 @@ if __name__ == "__main__":
     app = QApplication(sys.argv)
     apply_theme("light")
     dispatcher = BECDispatcher(gui_id="restart_server")
-    win = TemperatureController()
+    win = TemperatureControllerControl(config_only=True)
     win.show()
     sys.exit(app.exec_())

debye_bec/devices/univ_temp_cont/univ_temp_cont.py

@@ -2,13 +2,14 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Literal
 
 from ophyd import Component as Cpt
 from ophyd import EpicsSignal, EpicsSignalRO
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
+from pydantic import BaseModel
 
-from .univ_temp_cont_enums import UnivTempContSetpointSelect
+from .univ_temp_cont_enums import POWER_MAP, RAMP_RATE_MAP, UnivTempContSetpointSelect
 
 if TYPE_CHECKING:
     from bec_lib.devicemanager import ScanInfo
@@ -16,9 +17,20 @@ if TYPE_CHECKING:
 TIMEOUT_FOR_PV = 5
 
 
+class ConfigDict(BaseModel):
+    """A configuration dictionary."""
+
+    power_mode: Literal[tuple(POWER_MAP)]  # type: ignore[valid-type]
+    target_setpoint: float
+    ramp_rate: float
+    ramp_rate_unit: Literal[tuple(RAMP_RATE_MAP)]  # type: ignore[valid-type]
+
+
 class UnivTempCont(PSIDeviceBase):
     """Class for the Universal Temperature Controller"""
 
+    USER_ACCESS = ["load_config"]
+
     temperature = Cpt(
         EpicsSignalRO, suffix="TEMPERATURE", kind="normal", auto_monitor=True, doc="Temperature"
     )
@@ -81,3 +93,26 @@ class UnivTempCont(PSIDeviceBase):
             self.setpoint_select.set(UnivTempContSetpointSelect.REMOTE).wait(
                 timeout=self.timeout_for_pvwait
             )
+
+    def load_config(
+        self,
+        config: ConfigDict | None = None,
+        power_mode: str = "",
+        ramp_rate_unit: str = "",
+        ramp_rate: float = 0,
+        target_setpoint: float = 0,
+    ) -> None:
+        if config is None:
+            config = ConfigDict(
+                power_mode=power_mode,
+                ramp_rate_unit=ramp_rate_unit,
+                ramp_rate=ramp_rate,
+                target_setpoint=target_setpoint,
+            )
+        # Load a config to the controller (power, setpoint and ramp rate)
+        self.power.set(POWER_MAP[config.power_mode]).wait(timeout=self.timeout_for_pvwait)
+        self.ramp_rate_unit.set(RAMP_RATE_MAP[config.ramp_rate_unit]).wait(
+            timeout=self.timeout_for_pvwait
+        )
+        self.ramp_rate.set(config.ramp_rate).wait(timeout=self.timeout_for_pvwait)
+        self.target_setpoint.set(config.target_setpoint).wait(timeout=self.timeout_for_pvwait)