Source code for csaxs_bec.bec_ipython_client.plugins.flomni.gui_tools

import builtins
import time

from bec_lib import bec_logger

# from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_get, epics_put, fshopen, fshclose

logger = bec_logger.logger

if builtins.__dict__.get("bec") is not None:
    bec = builtins.__dict__.get("bec")
    dev = builtins.__dict__.get("dev")
    scans = builtins.__dict__.get("scans")


[docs] def umv(*args): return scans.umv(*args, relative=False)
[docs] class flomniGuiToolsError(Exception): pass
[docs] class flomniGuiTools: def __init__(self): self.text_box = None self.progressbar = None self.flomni_window = None self.xeyegui = None self.pdf_viewer = None self.idle_text_box = None self.camera_gripper_image = None self.camera_overview_image = None def set_client(self, client): self.client = client self.gui = self.client.gui def flomnigui_show_gui(self): if "flomni" in self.gui.windows: self.flomni_window = self.gui.windows["flomni"] self.gui.flomni.raise_window() else: self.flomni_window = self.gui.new("flomni") time.sleep(1) def flomnigui_stop_gui(self): self.gui.flomni.hide() def flomnigui_raise(self): self.gui.flomni.raise_window() def flomnigui_show_xeyealign(self): self.flomnigui_show_gui() if self._flomnigui_is_missing("xeyegui"): self.flomnigui_remove_all_docks() self.xeyegui = self.gui.flomni.new("XRayEye", object_name="xrayeye") # start live if not dev.cam_xeye.live_mode_enabled.get(): dev.cam_xeye.live_mode_enabled.put(True) self.xeyegui.switch_tab("alignment") def flomnigui_show_xeyealign_fittab(self): self.flomnigui_show_gui() if self._flomnigui_is_missing("xeyegui"): self.flomnigui_remove_all_docks() self.xeyegui = self.gui.flomni.new("XRayEye", object_name="xrayeye") self.xeyegui.switch_tab("fit") def _flomnigui_check_attribute_not_exists(self, attribute_name): if hasattr(self.gui, "flomni"): if attribute_name == "xeyegui": if hasattr(self.gui.flomni, "xrayeye"): return False if attribute_name == "progressbar": if hasattr(self.gui.flomni, "RingProgressBar"): return False if attribute_name == "cam_flomni_gripper" or attribute_name == "cam_flomni_overview": if hasattr(self.gui.flomni, "Image") or hasattr(self.gui.flomni, "Image_0"): return False return True def _flomnigui_is_missing(self, attribute_name): widget = getattr(self, attribute_name, None) if widget is None: return True if hasattr(widget, "_is_deleted") and widget._is_deleted(): return True return False def flomnigui_show_cameras(self): self.flomnigui_show_gui() if self._flomnigui_check_attribute_not_exists( "cam_flomni_gripper" ) or self._flomnigui_check_attribute_not_exists("cam_flomni_overview"): self.flomnigui_remove_all_docks() self.camera_gripper_image = self.gui.flomni.new("Image") if self._flomnicam_check_device_exists(dev.cam_flomni_gripper): self.camera_gripper_image.image(device="cam_flomni_gripper", signal="preview") self.camera_gripper_image.lock_aspect_ratio = True self.camera_gripper_image.enable_fps_monitor = True self.camera_gripper_image.enable_toolbar = False self.camera_gripper_image.outer_axes = False self.camera_gripper_image.inner_axes = False dev.cam_flomni_gripper.start_live_mode() else: print("Cannot open camera_gripper. Device does not exist.") self.camera_overview_image = self.gui.flomni.new("Image") if self._flomnicam_check_device_exists(dev.cam_flomni_overview): self.camera_overview_image.image(device="cam_flomni_overview", signal="preview") self.camera_overview_image.lock_aspect_ratio = True self.camera_overview_image.enable_fps_monitor = True self.camera_overview_image.enable_toolbar = False self.camera_overview_image.outer_axes = False self.camera_overview_image.inner_axes = False dev.cam_flomni_overview.start_live_mode() else: print("Cannot open camera_overview. Device does not exist.") def flomnigui_remove_all_docks(self): # dev.cam_flomni_overview.stop_live_mode() # dev.cam_flomni_gripper.stop_live_mode() # dev.cam_xeye.live_mode = False if hasattr(self.gui, "flomni"): self.gui.flomni.delete_all() self.progressbar = None self.text_box = None self.xeyegui = None self.pdf_viewer = None self.idle_text_box = None self.camera_gripper_image = None self.camera_overview_image = None def flomnigui_idle(self): self.flomnigui_show_gui() if self._flomnigui_is_missing("idle_text_box"): self.flomnigui_remove_all_docks() self.idle_text_box = self.gui.flomni.new("TextBox") text = ( "<pre>" + "██████╗ ███████╗ ██████╗ ███████╗██╗ ██████╗ ███╗ ███╗███╗ ██╗██╗\n" + "██╔══██╗██╔════╝██╔════╝ ██╔════╝██║ ██╔═══██╗████╗ ████║████╗ ██║██║\n" + "██████╔╝█████╗ ██║ █████╗ ██║ ██║ ██║██╔████╔██║██╔██╗ ██║██║\n" + "██╔══██╗██╔══╝ ██║ ██╔══╝ ██║ ██║ ██║██║╚██╔╝██║██║╚██╗██║██║\n" + "██████╔╝███████╗╚██████╗ ██║ ███████╗╚██████╔╝██║ ╚═╝ ██║██║ ╚████║██║\n" + "╚═════╝ ╚══════╝ ╚═════╝ ╚═╝ ╚══════╝ ╚═════╝ ╚═╝ ╚═╝╚═╝ ╚═══╝╚═╝\n" + "</pre>" ) self.idle_text_box.set_html_text(text) def flomnigui_docs(self, filename: str | None = None): import csaxs_bec from pathlib import Path print( "The general flOMNI documentation is at \nhttps://sls-csaxs.readthedocs.io/en/latest/user/ptychography/flomni.html#user-ptychography-flomni" ) csaxs_bec_basepath = Path(csaxs_bec.__file__).parent docs_folder = csaxs_bec_basepath / "bec_ipython_client" / "plugins" / "flomni" / "docs" if not docs_folder.is_dir(): raise NotADirectoryError(f"Docs folder not found: {docs_folder}") pdfs = sorted(docs_folder.glob("*.pdf")) if not pdfs: raise FileNotFoundError(f"No PDF files found in {docs_folder}") # --- Resolve PDF ------------------------------------------------------ if filename is not None: pdf_file = docs_folder / filename if not pdf_file.exists(): raise FileNotFoundError(f"Requested file not found: {filename}") else: print("\nAvailable flOMNI documentation PDFs:\n") for i, pdf in enumerate(pdfs, start=1): print(f" {i:2d}) {pdf.name}") print() while True: try: choice = int(input(f"Select a file (1–{len(pdfs)}): ")) if 1 <= choice <= len(pdfs): pdf_file = pdfs[choice - 1] break print(f"Enter a number between 1 and {len(pdfs)}.") except ValueError: print("Invalid input. Please enter a number.") # --- GUI handling (active existence check) ---------------------------- self.flomnigui_show_gui() if self._flomnigui_is_missing("pdf_viewer"): self.flomnigui_remove_all_docks() self.pdf_viewer = self.gui.flomni.new(widget="PdfViewerWidget") # --- Load PDF --------------------------------------------------------- self.pdf_viewer.load_pdf(str(pdf_file.resolve())) print(f"\nLoaded: {pdf_file.name}\n") def _flomnicam_check_device_exists(self, device): try: device except: return False else: return True def flomnigui_show_progress(self): try: self.flomnigui_show_gui() if self._flomnigui_is_missing("progressbar"): self.flomnigui_remove_all_docks() # Add a new dock with a RingProgressBar widget self.progressbar = self.gui.flomni.new("RingProgressBar") # Setting multiple rings with different values self.progressbar.add_ring().set_update("manual") self.progressbar.add_ring().set_update("manual") self.progressbar.add_ring().set_update("scan") self._flomnigui_update_progress() except Exception as e: logger.warning(f"Error in flomnigui_show_progress: {e}") def _flomnigui_update_progress(self): """Update the progress ring bar and center label from the current progress state. ``self.progress`` is backed by the BEC global variable ``tomo_progress`` (see :class:`_ProgressProxy` in ``flomni.py``), so this method reflects the live state that is also accessible from other BEC client sessions via:: client.get_global_var("tomo_progress") """ try: if self.progressbar is None: logger.warning("flOMNI progress GUI update skipped: progressbar is not available.") return main_progress_ring = self.progressbar.rings[0] subtomo_progress_ring = self.progressbar.rings[1] progress = self.progress["projection"] / self.progress["total_projections"] * 100 subtomo_progress = ( self.progress["subtomo_projection"] / self.progress["subtomo_total_projections"] * 100 ) main_progress_ring.set_value(progress) subtomo_progress_ring.set_value(subtomo_progress) # --- format start time for display -------------------------------- start_str = self.progress.get("tomo_start_time") if start_str is not None: import datetime as _dt start_display = _dt.datetime.fromisoformat(start_str).strftime("%Y-%m-%d %H:%M:%S") else: start_display = "N/A" # --- format estimated remaining time ------------------------------ remaining_s = self.progress.get("estimated_remaining_time") if remaining_s is not None and remaining_s >= 0: import datetime as _dt remaining_s = int(remaining_s) h, rem = divmod(remaining_s, 3600) m, s = divmod(rem, 60) if h > 0: eta_display = f"{h}h {m:02d}m {s:02d}s" elif m > 0: eta_display = f"{m}m {s:02d}s" else: eta_display = f"{s}s" else: eta_display = "N/A" # ------------------------------------------------------------------ # --- format estimated finish (wall-clock) time -------------------- finish_str = self.progress.get("estimated_finish_time") if finish_str is not None: import datetime as _dt finish_display = _dt.datetime.fromisoformat(finish_str).strftime( "%Y-%m-%d %H:%M:%S" ) else: finish_display = "N/A" # ------------------------------------------------------------------ text = ( f"Progress report:\n" f" Tomo type: {self.progress['tomo_type']}\n" f" Projection: {self.progress['projection']:.0f}\n" f" Total projections expected {self.progress['total_projections']:.1f}\n" f" Angle: {self.progress['angle']:.1f}\n" f" Current subtomo: {self.progress['subtomo']}\n" f" Current projection within subtomo: {self.progress['subtomo_projection']}\n" f" Total projections per subtomo: {int(self.progress['subtomo_total_projections'])}\n" f" Scan started: {start_display}\n" f" Est. remaining: {eta_display}\n" f" Est. finish: {finish_display}" ) self.progressbar.set_center_label(text) except Exception as exc: logger.warning(f"flOMNI progress GUI update skipped: {exc}")
if __name__ == "__main__": from bec_lib.client import BECClient from bec_widgets.cli.client_utils import BECGuiClient client = BECClient() client.start() client.gui = BECGuiClient() flomni_gui = flomniGuiTools() flomni_gui.set_client(client) flomni_gui.flomnigui_show_gui() flomni_gui.flomnigui_show_progress()