removed bl chk to replace by states

csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py

@@ -12,7 +12,6 @@ from bec_lib.pdf_writer import PDFWriter
 from typeguard import typechecked
 
 from csaxs_bec.bec_ipython_client.plugins.omny.omny_general_tools import (
-    BeamlineChecker,
     OMNYTools,
     PtychoReconstructor,
     TomoIDManager,
@@ -40,7 +39,6 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
         self.init = LaMNIInitStages(client)
 
         # Extracted collaborators
-        self.bl_chk = BeamlineChecker(client)
         self.reconstructor = PtychoReconstructor(self.ptycho_reconstruct_foldername)
         self.tomo_id_manager = TomoIDManager()
         self.OMNYTools = OMNYTools(self.client)
@@ -538,7 +536,6 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
             print(f"Starting LamNI scan for angle {angle} in subtomo {subtomo_number}")
             self._print_progress()
             while not successful:
-                self.bl_chk._bl_chk_start()
                 if not self.special_angles:
                     self._current_special_angles = []
                 if self._current_special_angles:
@@ -565,10 +562,9 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
                 for scan_nr in range(start_scan_number, end_scan_number):
                     self._write_tomo_scan_number(scan_nr, angle, subtomo_number)
 
-                if self.bl_chk._bl_chk_stop() and not error_caught:
+                #todo here bl chk
                     successful = True
-                else:
-                    self.bl_chk._bl_chk_wait_until_recovered()
+
 
     def _golden(self, ii, howmany_sorted, maxangle=360, reverse=False):
         """Return the ii-th golden ratio angle within sorted bunches and its subtomo number."""

csaxs_bec/bec_ipython_client/plugins/omny/omny_general_tools.py

@@ -2,6 +2,7 @@ import builtins
 import datetime
 import fcntl
 import os
+import socket
 import subprocess
 import sys
 import termios
@@ -10,6 +11,7 @@ import time
 import tty
 from pathlib import Path
 
+import epics
 import numpy as np
 from bec_lib import bec_logger
 from rich import box
@@ -156,199 +158,6 @@ class OMNYTools:
 
 import socket
 
-class BeamlineChecker:
-    """Monitors beamline health during scans.
-
-    Runs checks in a background thread and blocks scan progress
-    until beam conditions are restored if they fail.
-
-    Usage:
-        checker = BeamlineChecker(client)
-        checker._bl_chk_start()
-        # ... run scan ...
-        beam_was_ok = checker._bl_chk_stop()
-        if not beam_was_ok:
-            checker._bl_chk_wait_until_recovered()
-    """
-
-    def __init__(self, client):
-        self.client = client
-        self.check_shutter = True
-        self.check_light_available = True
-        self.check_fofb = True
-        self._beam_is_okay = True
-        self._stop_event = None
-        self._thread = None
-        self._local_network_warned = False
-        self._check_msgs = []
-
-    # ------------------------------------------------------------------
-    # Public control interface
-    # ------------------------------------------------------------------
-
-    def bl_chk_status(self):
-        """Print and return the current enabled/disabled state of all checks."""
-        if self._is_local_network():
-            print("Beamline checks cannot be performed on this network (129.129.98.x) — skipping.")
-            return {}
-        status = {
-            "shutter": self.check_shutter,
-            "fofb": self.check_fofb,
-            "light available": self.check_light_available,
-        }
-        print(
-            f"Shutter: {self.check_shutter}\n"
-            f"FOFB: {self.check_fofb}\n"
-            f"Light available: {self.check_light_available}"
-        )
-        return status
-
-    def bl_chk_enable_all(self):
-        """Enable all beamline checks."""
-        self.check_shutter = True
-        self.check_light_available = True
-        self.check_fofb = True
-        self.bl_chk_status()
-
-    def bl_chk_disable_all(self):
-        """Disable all beamline checks."""
-        self.check_shutter = False
-        self.check_light_available = False
-        self.check_fofb = False
-        self.bl_chk_status()
-
-    def bl_chk_enable_shutter(self):
-        """Enable the shutter check."""
-        self.check_shutter = True
-        self.bl_chk_status()
-
-    def bl_chk_disable_shutter(self):
-        """Disable the shutter check."""
-        self.check_shutter = False
-        self.bl_chk_status()
-
-    def bl_chk_enable_fofb(self):
-        """Enable the fast orbit feedback check."""
-        self.check_fofb = True
-        self.bl_chk_status()
-
-    def bl_chk_disable_fofb(self):
-        """Disable the fast orbit feedback check."""
-        self.check_fofb = False
-        self.bl_chk_status()
-
-    def bl_chk_enable_light(self):
-        """Enable the light available check."""
-        self.check_light_available = True
-        self.bl_chk_status()
-
-    def bl_chk_disable_light(self):
-        """Disable the light available check."""
-        self.check_light_available = False
-        self.bl_chk_status()
-
-    def _bl_chk_start(self):
-        """Start the background beam check thread."""
-        self._beam_is_okay = True
-        self._stop_event = threading.Event()
-        self._thread = threading.Thread(target=self._poll, daemon=True)
-        self._thread.start()
-
-    def _bl_chk_stop(self) -> bool:
-        """Stop the background thread and return whether beam was okay throughout."""
-        self._stop_event.set()
-        self._thread.join()
-        return self._beam_is_okay
-
-    def _bl_chk_wait_until_recovered(self):
-        """Block until all beamline checks pass again, logging to SciLog."""
-        self._log_failure_to_scilog()
-        while True:
-            self._beam_is_okay = True
-            self._check_msgs = self._run_checks()
-            if self._beam_is_okay:
-                break
-            self._print_msgs()
-            time.sleep(1)
-        self._log_recovery_to_scilog()
-
-    # ------------------------------------------------------------------
-    # Internal
-    # ------------------------------------------------------------------
-
-    def _is_local_network(self) -> bool:
-        """Return True if running on the 129.129.98.x subnet."""
-        try:
-            hostname = socket.gethostname()
-            ip = socket.gethostbyname(hostname)
-            return ip.startswith("129.129.98.")
-        except Exception:
-            return False
-
-    def _run_checks(self) -> list:
-        if self._is_local_network():
-            if not self._local_network_warned:
-                print("Beamline checks cannot be performed on this network (129.129.98.x) — skipping.")
-                self._local_network_warned = True
-            return []
-        msgs = []
-        dev = builtins.__dict__.get("dev")
-        try:
-            if self.check_shutter:
-                val = dev.x12sa_es1_shutter_status.read(cached=True)
-                if val["value"].lower() != "open":
-                    self._beam_is_okay = False
-                    msgs.append("Check beam failed: Shutter is closed.")
-            if self.check_light_available:
-                val = dev.sls_machine_status.read(cached=True)
-                if val["value"] not in ["Light Available", "Light-Available"]:
-                    self._beam_is_okay = False
-                    msgs.append("Check beam failed: Light not available.")
-            if self.check_fofb:
-                val = dev.sls_fast_orbit_feedback.read(cached=True)
-                if val["value"] != "running":
-                    self._beam_is_okay = False
-                    msgs.append("Check beam failed: Fast orbit feedback is not running.")
-        except Exception:
-            logger.warning("Failed to check beam.")
-        return msgs
-
-    def _poll(self):
-        while not self._stop_event.is_set():
-            self._check_msgs = self._run_checks()
-            if not self._beam_is_okay:
-                self._stop_event.set()
-            time.sleep(1)
-
-    def _print_msgs(self):
-        for msg in self._check_msgs:
-            logger.warning(msg)
-
-    def _log_failure_to_scilog(self):
-        self._print_msgs()
-        try:
-            bec = builtins.__dict__.get("bec")
-            msg = bec.logbook.LogbookMessage()
-            msg.add_text(
-                "<p><mark class='pen-red'><strong>Beamline checks failed at"
-                f" {str(datetime.datetime.now())}: {''.join(self._check_msgs)}</strong></mark></p>"
-            ).add_tag(["BEC", "beam_check"])
-            self.client.logbook.send_logbook_message(msg)
-        except Exception:
-            logger.warning("Failed to send beam failure update to SciLog.")
-
-    def _log_recovery_to_scilog(self):
-        try:
-            bec = builtins.__dict__.get("bec")
-            msg = bec.logbook.LogbookMessage()
-            msg.add_text(
-                "<p><mark class='pen-red'><strong>Operation resumed at"
-                f" {str(datetime.datetime.now())}.</strong></mark></p>"
-            ).add_tag(["BEC", "beam_check"])
-            self.client.logbook.send_logbook_message(msg)
-        except Exception:
-            logger.warning("Failed to send beam recovery update to SciLog.")
-
 class PtychoReconstructor:
     """Writes ptychography reconstruction queue files after each scan projection.
 
@@ -389,6 +198,10 @@ class PtychoReconstructor:
                                     name the queue file.
             base_path (str): Root path under which the queue folder lives.
         """
+        if not self._accounts_match():
+            logger.warning("Active BEC account does not match system user — skipping queue file write.")
+            return
+        
         base_path = os.path.expanduser(base_path)
         queue_path = Path(os.path.join(base_path, self.folder_name))
         queue_path.mkdir(parents=True, exist_ok=True)