wip digital twin

debye_bec/bec_ipython_client/plugins/digital_twin_core/motion_worker.py

@@ -244,10 +244,12 @@ class MotionWorker:
                     self.dev[self.motor].stop()
                     self._stop_flag.clear()
                 if rb is not None:
-                    self._emit_position_changed(rb["device"].read(cached=True)[rb["name"]]["value"])
+                    self._emit_position_changed(
+                        rb["device"].read(cached=False)[rb["name"]]["value"]
+                    )
                 else:
                     self._emit_position_changed(
-                        self.dev[self.motor].read(cached=True)[self.motor]["value"]
+                        self.dev[self.motor].read(cached=False)[self.motor]["value"]
                     )
                 if surveyed_axes is not None:
                     for surv_ax in surveyed_axes:

debye_bec/bec_ipython_client/plugins/digital_twin_core/x01da_parameters.py

@@ -320,4 +320,4 @@ vacuum_pipes = pipe(
 )
 
 Walls = namedtuple("walls", ["start", "end", "height"])
-walls = Walls(start=[13999.30], end=[13999 + 75.5 + 30], height=[[-20, 25]])
+walls = Walls(start=[13999.30], end=[13999 + 75.5 + 30], height=[[-20, 25]])  # WBBS

debye_bec/bec_ipython_client/plugins/digital_twin_core/x10da_parameters.py

@@ -294,3 +294,6 @@ ic = namedtuple("sample", ["name", "center"])
 es1ic0 = ic(name="ES1 IC0", center=[0, 23633, sourceHeight])
 es1ic1 = ic(name="ES1 IC1", center=[0, 24383, sourceHeight])
 es1ic2 = ic(name="ES1 IC2", center=[0, 24723, sourceHeight])
+
+Walls = namedtuple("walls", ["start", "end", "height"])
+walls = Walls(start=[13472], end=[13472 + 354], height=[[-20, 36]])  # WBBS

debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py

@@ -2,9 +2,13 @@
 Digital Twin: Custom BEC widget to support the beamline alignment.
 """
 
+import json
 import sys
+import time
+from pathlib import Path
 from typing import Literal, cast
 
+import h5py
 import numpy as np
 import yaml
 from bec_lib import bec_logger
@@ -22,9 +26,11 @@ from qtpy.QtWidgets import (
     QComboBox,
     QDialog,
     QDialogButtonBox,
+    QFileDialog,
     QFrame,
     QHBoxLayout,
     QLabel,
+    QMessageBox,
     QPlainTextEdit,
     QPushButton,
     QScrollArea,
@@ -141,6 +147,9 @@ class DigitalTwin(BECWidget, QWidget):
         self.input.adapt_reality.clicked_connect(self.adapt_reality)
         self.settings.load_offsets.clicked_connect(self.load_offsets)
         self.settings.show_offsets.clicked_connect(self.show_offsets)
+        self.settings.save_config.clicked_connect(self.save_config)
+        self.settings.load_from_config.clicked_connect(self.load_from_config)
+        self.settings.load_from_scan.clicked_connect(self.load_from_scan)
 
         self.edge_selector_energy = 0.0
         self.bragg_angle = 0.0
@@ -506,7 +515,7 @@ class DigitalTwin(BECWidget, QWidget):
             mo1_mode = "Monochromatic"
         else:
             mo1_mode = "Pinkbeam"
-        mo1_bragg = self.dev.mo1_bragg.read(cached=True)
+        mo1_bragg = self.dev.mo1_bragg.read(cached=False)
         sldi_gapx = self.dev.sldi_gapx.read(cached=True)["sldi_gapx"]["value"]
         sldi_gapy = self.dev.sldi_gapy.read(cached=True)["sldi_gapy"]["value"]
         h_acc, v_acc = self.core.sldi_gap_to_acc(sldi_gapx, sldi_gapy)
@@ -647,7 +656,7 @@ class DigitalTwin(BECWidget, QWidget):
         # Removing offsets
         pos = self.core.remove_offsets(pos)
 
-        self.input.energy.set_number(self.dev.mo1_bragg.read(cached=True)["mo1_bragg"]["value"])
+        self.input.energy.set_number(self.dev.mo1_bragg.read(cached=False)["mo1_bragg"]["value"])
         h_acc, v_acc = self.core.sldi_gap_to_acc(pos["sldi_gapx"], pos["sldi_gapy"])
         self.input.sldi_hacc.set_number(h_acc * 1e3)
         self.input.sldi_vacc.set_number(v_acc * 1e3)
@@ -659,7 +668,7 @@ class DigitalTwin(BECWidget, QWidget):
             mo1_mode = "Pinkbeam"
         self.input.mo1_mode.set_current_text(mo1_mode)
         self.input.mo1_xtal.set_current_text(
-            self.dev.mo1_bragg.read(cached=True)["mo1_bragg_crystal_current_xtal_string"]["value"]
+            self.dev.mo1_bragg.read(cached=False)["mo1_bragg_crystal_current_xtal_string"]["value"]
         )
         fm_stripe = self.core.fm_trx_to_stripe(-pos["fm_trx"])
         self.input.fm_stripe.set_current_text(fm_stripe)
@@ -798,6 +807,166 @@ class DigitalTwin(BECWidget, QWidget):
 
         dialog.exec()
 
+    @SafeSlot()
+    def save_config(self, *_):
+        config = self.get_assistant_config()
+
+        hostname = self.client._hostname
+        start = hostname.find("x")
+        if start != -1:
+            beamline = hostname[start : start + 5]
+            active_account = self.client.active_account
+            folder = f"/sls/{beamline}/data/{active_account}/raw/widgets/digital_twin/configs"
+        else:
+            folder = "/sls"
+
+        folder = Path(folder)
+        folder.mkdir(parents=True, exist_ok=True)
+
+        t = time.localtime()
+        fmt_time = time.strftime("%Y%m%d_%H%M%S", t)
+        default_filename = "config_" + fmt_time + ".json"
+
+        filename, _ = QFileDialog.getSaveFileName(
+            self,
+            "Save dictionary",
+            str(folder / default_filename),
+            "JSON files (*.json);;All files (*)",
+        )
+
+        if not filename:
+            return None
+
+        path = Path(filename)
+
+        if path.suffix.lower() != ".json":
+            path = path.with_suffix(".json")
+
+        with path.open("w", encoding="utf-8") as f:
+            json.dump(config, f, indent=4, ensure_ascii=False)
+
+        return path
+
+    @SafeSlot()
+    def load_from_config(self, *_):
+        hostname = self.client._hostname
+        start = hostname.find("x")
+        if start != -1:
+            beamline = hostname[start : start + 5]
+            active_account = self.client.active_account
+            folder = f"/sls/{beamline}/data/{active_account}/raw/widgets/digital_twin/configs"
+        else:
+            folder = "/sls"
+
+        path, _ = QFileDialog.getOpenFileName(
+            self, "Load config from file", folder, "JSON files (*.json);;All files (*)"
+        )
+        if not path:
+            return
+
+        try:
+            with open(path, encoding="utf-8") as f:
+                config = json.load(f)
+        except (OSError, json.JSONDecodeError) as exc:
+            QMessageBox.critical(
+                self, "Load failed", f"Could not load a config from {path}:\n{exc}"
+            )
+            return
+
+        self.set_assistant_config(config)
+
+    @SafeSlot()
+    def load_from_scan(self, *_):
+        hostname = self.client._hostname
+        start = hostname.find("x")
+        if start != -1:
+            beamline = hostname[start : start + 5]
+            active_account = self.client.active_account
+            start_folder = f"/sls/{beamline}/data/{active_account}/raw"
+        else:
+            start_folder = "/sls"
+
+        filename, _ = QFileDialog.getOpenFileName(
+            self, "Open scan", start_folder, "h5 files (*.h5);;All files (*)"
+        )
+
+        if not filename:
+            return None, None
+
+        path = Path(filename)
+
+        with h5py.File(path, "r") as f:
+            devices = cast(h5py.Group, f["entry/collection/devices"])
+
+            pos = {}
+            # fmt:off
+            pos["sldi_gapx"] = cast(h5py.Dataset, devices["sldi_gapx/sldi_gapx/value"])[()]
+            pos["sldi_gapy"] = cast(h5py.Dataset, devices["sldi_gapy/sldi_gapy/value"])[()]
+            pos["cm_trx"] = cast(h5py.Dataset, devices["cm_trx/cm_trx/value"])[()]
+            pos["cm_rotx"] = cast(h5py.Dataset, devices["cm_rotx/cm_rotx/value"])[()]
+            pos["mo1_trx"] = cast(h5py.Dataset, devices["mo1_trx/mo1_trx/value"])[()]
+            pos["fm_trx"] = cast(h5py.Dataset, devices["fm_trx/fm_trx/value"])[()]
+            pos["fm_rotx"] = cast(h5py.Dataset, devices["fm_rotx/fm_rotx/value"])[()]
+            pos["fm_bnd_radius"] = cast(h5py.Dataset, devices["fm_bnd_radius/fm_bnd_radius/value"])[()]
+            pos["es1man_trz"] = cast(h5py.Dataset, devices["es1man_trz/es1man_trz/value"])[()]
+            if self.beamline == BeamlineId.X01DA:
+                pos["ot_es1_trz"] = cast(h5py.Dataset, devices["ot_es1_trz/ot_es1_trz/value"])[()]
+                pos["ot_es2_trz"] = cast(h5py.Dataset, devices["ot_es2_trz/ot_es2_trz/value"])[()]
+                pos["smpl_to_xrd"] = pos["ot_es2_trz"] - pos["ot_es1_trz"] - pos["es1man_trz"] + 32
+            else:
+                table = self.ask_table_selection(self.input.table.currentText())
+                pos["ot_es1_trz"] = self.core.table_to_smpl_pos(table)
+            # fmt:on
+
+            # Removing offsets
+            pos = self.core.remove_offsets(pos)
+
+            self.input.energy.set_number(
+                cast(h5py.Dataset, devices["mo1_bragg/mo1_bragg/value"])[()]
+            )
+            h_acc, v_acc = self.core.sldi_gap_to_acc(pos["sldi_gapx"], pos["sldi_gapy"])
+            self.input.sldi_hacc.set_number(h_acc * 1e3)
+            self.input.sldi_vacc.set_number(v_acc * 1e3)
+            self.input.cm_stripe.set_current_text(self.core.cm_trx_to_stripe(-pos["cm_trx"]))
+            self.input.cm_pitch.set_number(pos["cm_rotx"])
+            if abs(pos["mo1_trx"]) > 5:
+                mo1_mode = "Monochromatic"
+            else:
+                mo1_mode = "Pinkbeam"
+            self.input.mo1_mode.set_current_text(mo1_mode)
+            self.input.mo1_xtal.set_current_text(
+                cast(
+                    h5py.Dataset, devices["mo1_bragg/mo1_bragg_crystal_current_xtal_string/value"]
+                )[()]
+            )
+
+        fm_stripe = self.core.fm_trx_to_stripe(-pos["fm_trx"])
+        self.input.fm_stripe.set_current_text(fm_stripe)
+        fm_rotx_real = 2 * pos["cm_rotx"] - pos["fm_rotx"]
+        logger.info(f"In adapt reality, fm_rotx_real is {fm_rotx_real}")
+        self.input.fm_rotx.set_number(fm_rotx_real)
+
+        if self.beamline == BeamlineId.X01DA:
+            self.input.ot_es1_trz.set_number(pos["ot_es1_trz"])
+            self.input.smpl_to_xrd.set_number(pos["smpl_to_xrd"])
+        else:
+            table = self.ask_table_selection(self.input.table.currentText())
+            self.input.table.set_current_text(table)
+
+        smpl = pos["ot_es1_trz"] + pos["es1man_trz"]
+
+        fm_focx, fm_focy = self.core.calc_beamsize(
+            h_acc, v_acc, fm_stripe, -fm_rotx_real * 1e-3, pos["fm_bnd_radius"] * 1e6, smpl
+        )
+        if fm_focx < 0.08 and fm_focy < 0.08:
+            self.input.fm_focus.set_current_text("Focused")
+        else:
+            self.input.fm_focus.set_current_text("Defocused")
+            self.input.fm_focx.set_number(fm_focx)
+            self.input.fm_focy.set_number(fm_focy)
+
+        self.calc_assistant(identifier="init")
+
     def update_fm_mode(self):
         """
         Updates the focusing mirror input group based on the
@@ -972,11 +1141,11 @@ class DigitalTwin(BECWidget, QWidget):
         """
         xtal = self.input.mo1_xtal.currentText()
         if xtal == "Si(111)":
-            d_spacing = self.dev.mo1_bragg.crystal.d_spacing_si111.read(cached=True)[
+            d_spacing = self.dev.mo1_bragg.crystal.d_spacing_si111.read(cached=False)[
                 "mo1_bragg_crystal_d_spacing_si111"
             ]["value"]
         elif xtal == "Si(311)":
-            d_spacing = self.dev.mo1_bragg.crystal.d_spacing_si311.read(cached=True)[
+            d_spacing = self.dev.mo1_bragg.crystal.d_spacing_si311.read(cached=False)[
                 "mo1_bragg_crystal_d_spacing_si311"
             ]["value"]
         else:

debye_bec/bec_widgets/widgets/digital_twin/panels/settings_panel.py

@@ -27,5 +27,20 @@ class SettingsPanel(QWidget):
             "Axes Offsets", [self.load_offsets, self.offsets_status, self.show_offsets]
         )
 
-        self._layout.addWidget(self.offset_group)
+        self.load_from_scan = Button(
+            label="Load from Scan", label_button="Load from Scan", enabled=True
+        )
+        self.load_from_config = Button(
+            label="Load from Config", label_button="Load from Config", enabled=True
+        )
+        self.save_config = Button(label="Save Config", label_button="Save Config", enabled=True)
+
+        self.import_export_group = Group(
+            "Import/Export", [self.load_from_scan, self.load_from_config, self.save_config]
+        )
+
+        # Assemble complete settings group
+        self.settings_group = Group("Settings", [self.offset_group, self.import_export_group])
+
+        self._layout.addWidget(self.settings_group)
         self._layout.addStretch()