global constant control

jungfrau_gui/globals.py

@@ -55,6 +55,7 @@ def get_git_info():
 # jfj = False
 
 tem_host = cfg.temserver
+tem_port = 3535
 dev = False
 #Configuration
 nrow = cfg.nrows 
@@ -87,8 +88,51 @@ def get_git_info():
 # constants, presets
 UM_TO_NM = 1e3
 MM_TO_UM = 1e3
+MS_TO_US = 1e3
+S_TO_MS = 1e3
 KV_TO_V = 1e3 
 PIXEL = 0.075 # mm
 
+# TEM control variables
 default_HT = 200000.00 # V
 backlash = [100, 80, 0, 0]
+
+min_mag_for_mag = 1500 # border between LowMag/Mag
+
+## safety not to hit hardware-limit
+click_on_move_thresholds = {'dxy_min': 0.3, 'dxy_max': 100, 
+                            'dz_min_mag': 1, 'dz_max_mag': 10,
+                            'dz_min_lmag': 3, 'absz_min': -70, 'absz_max': 20}
+
+## stage shift larger than these values will be hold in history
+stage_relaxation_thresholds = [30, 30, 30, 0.2, 100] # nm, nm, nm, deg., deg. 
+
+## software limit for GATAN holder. Smaller value (~65) may be necessary for complete safety (e.g. remote-operation).
+max_stage_tilt = 72
+default_roation_end = 60
+
+## variabls for autofocusing
+IL1_0 = 21780 # 21819 
+ILS_0 = [32920, 32776] # [32820, 32976]
+WAIT_TIME_S = 0.25 # TODO: optimize value
+
+
+# Frame control variables
+default_polling_frequency = 1000
+min_polling_frequency = 100 # safety not to inquire TEM-values too frequently
+max_polling_frequency = 10000
+
+default_frame_summed = 100
+default_image_time_us = 500
+min_frame_summed = 10 # safety not to save unexpectedly large datasets
+max_frame_summed = 1000
+detector_freq = 2000
+max_duration = 3600 # sec
+
+
+# Communication variables
+dataserver_host = "noether"
+dataserver_port = 3463
+
+max_retries_tagging = 3
+inquiry_delay = 0.1 # sec

jungfrau_gui/metadata_uploader/metadata_update_client.py

@@ -33,7 +33,7 @@ def default(self, obj):
         return super().default(obj)
 
 class MetadataNotifier:
-    def __init__(self, host, port=3463, verbose = True):
+    def __init__(self, host, port=globals.dataserver_port, verbose = True):
         self.host = host
         self.port = port
         self.verbose = verbose

jungfrau_gui/ui_components/file_operations/file_operations.py

@@ -81,7 +81,7 @@ def __init__(self, parent):
         self.cfg = ConfigurationClient(redis_host(), token=auth_token())
         self.trigger_update_h5_index_box.connect(self.update_index_box)
         self.initUI()
-        self.metadata_notifier = MetadataNotifier(host = "noether", port = 3463, verbose = False)
+        self.metadata_notifier = MetadataNotifier(host = globals.dataserver_host, port = globals.dataserver_port, verbose = False)
 
 
     def initUI(self):
@@ -421,11 +421,11 @@ def toggle_snapshot_btn(self):
             if globals.dev:
                 prev_image_time_us = self.parent.visualization_panel.jfjoch_client.image_time_us # 50000
                 self.frame_summed_for_rotation = self.parent.visualization_panel.frame_summed.value()
-                frame_summed_for_snapshot = prev_image_time_us // 500 # 100
+                frame_summed_for_snapshot = prev_image_time_us // globals.default_image_time_us # 100
                 self.parent.visualization_panel.frame_summed.setValue(frame_summed_for_snapshot)
 
             self.parent.visualization_panel.send_command_to_jfjoch('collect')
-            logging.info(f'Snapshot duration: {int(self.snapshot_spin.value())*1e-3} sec')
+            logging.info(f'Snapshot duration: {int(self.snapshot_spin.value())/globals.S_TO_MS} sec')
             QTimer.singleShot(self.snapshot_spin.value(), self.toggle_snapshot_btn)
         else:
             # Cancel collection
@@ -487,8 +487,8 @@ def _send_metadata_with_retries(self, beam_property):
                 beam_property,
                 None,  # self.rotations_angles,
                 self.cfg.threshold,
-                retries=3, 
-                delay=0.1
+                retries=globals.max_retries_tagging, 
+                delay=globals.inquiry_delay
             )
             logging.info("Metadata update completed successfully")
             # Signal success back to the main thread
@@ -523,7 +523,7 @@ def _disconnect_metadata_signals(self):
 
     def _finalize_snapshot(self):
         """Reset UI state after snapshot completion"""
-        logging.info(f'Snapshot duration end: {int(self.snapshot_spin.value())*1e-3} sec')
+        logging.info(f'Snapshot duration end: {int(self.snapshot_spin.value())/globals.S_TO_MS} sec')
         self.tag_input.setText(self.pre_text)  # reset the tag to value before snapshot
         self.update_measurement_tag()
         self.snapshot_button.setText("Write Stream as a snapshot-H5")

jungfrau_gui/ui_components/file_operations/processresult_updater.py

@@ -5,7 +5,7 @@
 from datetime import datetime
 import argparse
 from pathlib import Path
-# from .. import globals
+from ... import globals
 from PySide6.QtCore import Signal, Slot, QObject
 import time
 
@@ -26,7 +26,7 @@
 class ProcessedDataReceiver(QObject):
     finished = Signal()
 
-    def __init__(self, parent, host, port=3463, verbose = True, mode=0):
+    def __init__(self, parent, host, port=globals.dataserver_port, verbose = True, mode=0):
         super().__init__()
         self.task_name = "Processed Data Receiver"
         self.parent = parent
@@ -61,7 +61,7 @@ def run(self, timeout_ms = 5000, update_interval_ms=2000, n_retry=10, verbose =
                     socket.send_string("Results being inquired...")
                     result_json = socket.recv_string()
                     if 'In processing...' in result_json:
-                        time.sleep(update_interval_ms/1000)
+                        time.sleep(update_interval_ms/globals.S_TO_MS)
                         self.trial -= 1
                     elif 'Feedback is not activated.' in result_json:
                         logging.info("Server does not run in the feedback mode. Inquiry cloded.")
@@ -75,7 +75,7 @@ def run(self, timeout_ms = 5000, update_interval_ms=2000, n_retry=10, verbose =
                         break
                 except zmq.ZMQError as e:
                     logging.error(f"Failed to receive processed data request: {e}")
-                    time.sleep(update_interval_ms/1000)
+                    time.sleep(update_interval_ms/globals.S_TO_MS)
                     self.trial -= 1
                 # finally:
                 #     # ensure the socket is closed no matter what

jungfrau_gui/ui_components/tem_controls/connectivity_inspector.py

@@ -11,7 +11,7 @@ class TEM_Connector(QObject):
     def __init__(self):
         super(TEM_Connector, self).__init__()
         self.task_name = "TEM Connector"
-        self.client = TEMClient(globals.tem_host, 3535, verbose=False) 
+        self.client = TEMClient(globals.tem_host, globals.tem_port, verbose=False) 
 
     @Slot()
     def run(self):

jungfrau_gui/ui_components/tem_controls/task/beam_focus_task.py

@@ -10,10 +10,6 @@
 
 from simple_tem import TEMClient
 
-IL1_0 = 21780 # 21819 
-ILS_0 = [32920, 32776] # [32820, 32976]
-WAIT_TIME_S = 0.25 # TODO: optimize value
-
 class AutoFocusTask(Task):
     # Signal to notify the main thread that a new best result arrived
     newBestResult = Signal(dict)
@@ -24,14 +20,14 @@ def __init__(self, control_worker):
         self.estimateds_duration = self.duration_s + 0.1
         self.control = control_worker
         self.tem_action = self.control.tem_action
-        self.client = TEMClient(globals.tem_host, 3535)
+        self.client = TEMClient(globals.tem_host, globals.tem_port)
         # Start from a known set of lens values (but creates a freeze of ~0.1s) 
-        self.client.SetILFocus(IL1_0)
-        self.client.SetILs(*ILS_0)
-        time.sleep(WAIT_TIME_S)
+        self.client.SetILFocus(globals.IL1_0)
+        self.client.SetILs(*globals.ILS_0)
+        time.sleep(globals.WAIT_TIME_S)
         self.lens_parameters = {
-                                "il1": IL1_0, # an integer
-                                "ils": ILS_0, # two integers for stigmation
+                                "il1": globals.IL1_0, # an integer
+                                "ils": globals.ILS_0, # two integers for stigmation
         }
         self.beam_fitter = self.control.beam_fitter
         self.results = []
@@ -48,7 +44,7 @@ def __init__(self, control_worker):
         self._best_stigmation = None
         self._best_combined = None
 
-    def run(self, init_IL1=IL1_0, init_stigm=ILS_0, time_budget=15):
+    def run(self, init_IL1=globals.IL1_0, init_stigm=globals.ILS_0, time_budget=15):
         try:
             # ----------------------
             # Start parallel process
@@ -84,7 +80,7 @@ def run(self, init_IL1=IL1_0, init_stigm=ILS_0, time_budget=15):
     ##########
     # METHOD A
     ##########
-    def rapid_parabolic_focus(self, init_IL1, range_width=100, num_points=5, wait_time_s=WAIT_TIME_S):
+    def rapid_parabolic_focus(self, init_IL1, range_width=100, num_points=5, wait_time_s=globals.WAIT_TIME_S):
         """
         Rapidly optimize beam focus by sampling points and fitting a parabola.
 
@@ -248,7 +244,7 @@ def rapid_parabolic_focus(self, init_IL1, range_width=100, num_points=5, wait_ti
             else:
                 return None
 
-    def rapid_stigmation_optimization(self, init_stigm, deviation=100, num_points=10, wait_time_s=WAIT_TIME_S):
+    def rapid_stigmation_optimization(self, init_stigm, deviation=100, num_points=10, wait_time_s=globals.WAIT_TIME_S):
         """
         Rapidly optimize beam stigmation using a simplified grid search.
 
@@ -343,10 +339,10 @@ def rapid_autofocus(self, init_IL1=None, init_stigm=None, time_budget_seconds=15
         try:
             # Get current lens values if not provided
             if init_IL1 is None:
-                init_IL1 = self.lens_parameters.get("il1", IL1_0)
+                init_IL1 = self.lens_parameters.get("il1", globals.IL1_0)
 
             if init_stigm is None:
-                init_stigm = self.lens_parameters.get("ils", ILS_0)
+                init_stigm = self.lens_parameters.get("ils", globals.ILS_0)
 
             # Reset results and best tracking
             self.results = []
@@ -359,7 +355,7 @@ def rapid_autofocus(self, init_IL1=None, init_stigm=None, time_budget_seconds=15
                 init_IL1=init_IL1,
                 range_width=100,  # Adjust based on expected focus range
                 num_points=7,     # 7 points is a good balance between speed and accuracy ?
-                wait_time_s=WAIT_TIME_S   # Reduced wait time for speed
+                wait_time_s=globals.WAIT_TIME_S   # Reduced wait time for speed
             )
 
             if focus_result:
@@ -388,7 +384,7 @@ def rapid_autofocus(self, init_IL1=None, init_stigm=None, time_budget_seconds=15
                 init_stigm=init_stigm,
                 deviation=100,    # Adjust based on expected stigmation range
                 num_points=5,     # number of points in each dimension
-                wait_time_s=WAIT_TIME_S   # Reduced wait time for speed
+                wait_time_s=globals.WAIT_TIME_S   # Reduced wait time for speed
             )
 
             if stigmation_result:
@@ -409,7 +405,7 @@ def rapid_autofocus(self, init_IL1=None, init_stigm=None, time_budget_seconds=15
                     init_IL1=optimal_il1,
                     range_width=60,   # Narrower range for refinement
                     num_points=4,     # Fewer points needed for refinement
-                    wait_time_s=WAIT_TIME_S
+                    wait_time_s=globals.WAIT_TIME_S
                 )
 
                 if final_focus_result:
@@ -437,7 +433,7 @@ def rapid_autofocus(self, init_IL1=None, init_stigm=None, time_budget_seconds=15
     ##########
     # METHOD B
     ##########
-    def standard_focus(self, init_IL1=IL1_0, init_stigm=ILS_0):      
+    def standard_focus(self, init_IL1=globals.IL1_0, init_stigm=globals.ILS_0):      
             # Start counter
             autofocus_start = time.perf_counter()
 
@@ -552,7 +548,7 @@ def standard_focus(self, init_IL1=IL1_0, init_stigm=ILS_0):
             # Emit final result signal (for UI update)
             self.newBestResult.emit(final_results)
 
-    def sweep_il1_linear(self, lower, upper, step, wait_time_s=WAIT_TIME_S):
+    def sweep_il1_linear(self, lower, upper, step, wait_time_s=globals.WAIT_TIME_S):
         """
         Perform a linear sweep of IL1 TEM lens positions with Gaussian fitting at each step.
 
@@ -595,7 +591,7 @@ def sweep_il1_linear(self, lower, upper, step, wait_time_s=WAIT_TIME_S):
 
         return True
 
-    def sweep_stig_linear(self, init_stigm, deviation, step, wait_time_s=WAIT_TIME_S):
+    def sweep_stig_linear(self, init_stigm, deviation, step, wait_time_s=globals.WAIT_TIME_S):
         """
         Perform a linear sweep of stigmation parameters in X and Y directions.
 
@@ -702,15 +698,15 @@ def goto_il1_with_hysteresis_compensation(self, target_il1, margin=20):
 
         # Go to a value well below the target
         self.client.SetILFocus(target_il1 - 50)
-        time.sleep(WAIT_TIME_S)
+        time.sleep(globals.WAIT_TIME_S)
 
         # Overshoot by a fixed amount
         self.client.SetILFocus(target_il1 + margin)
-        time.sleep(WAIT_TIME_S)
+        time.sleep(globals.WAIT_TIME_S)
 
         # Approach the final value
         self.client.SetILFocus(target_il1)
-        time.sleep(WAIT_TIME_S)
+        time.sleep(globals.WAIT_TIME_S)
 
         # Update stored value
         self.lens_parameters["il1"] = target_il1
@@ -729,19 +725,19 @@ def goto_ils_with_hysteresis_compensation(self, target_ils, margin=50):
         # X axis approach
         current_ils_y = self.lens_parameters.get("ils", [0, 0])[1]
         self.client.SetILs(ils_x - 100, current_ils_y)
-        time.sleep(WAIT_TIME_S)
+        time.sleep(globals.WAIT_TIME_S)
         self.client.SetILs(ils_x + margin, current_ils_y)
-        time.sleep(WAIT_TIME_S)
+        time.sleep(globals.WAIT_TIME_S)
         self.client.SetILs(ils_x, current_ils_y)
-        time.sleep(WAIT_TIME_S)
+        time.sleep(globals.WAIT_TIME_S)
 
         # Y axis approach
         self.client.SetILs(ils_x, ils_y - 100)
-        time.sleep(WAIT_TIME_S)
+        time.sleep(globals.WAIT_TIME_S)
         self.client.SetILs(ils_x, ils_y + margin)
-        time.sleep(WAIT_TIME_S)
+        time.sleep(globals.WAIT_TIME_S)
         self.client.SetILs(ils_x, ils_y)
-        time.sleep(WAIT_TIME_S)
+        time.sleep(globals.WAIT_TIME_S)
 
         # Update stored value
         self.lens_parameters["ils"] = [ils_x, ils_y]

jungfrau_gui/ui_components/tem_controls/task/record_task.py

@@ -28,9 +28,9 @@ def __init__(self, control_worker, end_angle = 60, log_suffix = 'RotEDlog_test',
         self.rotations_angles = []
         self.log_suffix = log_suffix
         logging.info("RecordTask initialized")
-        self.client = TEMClient(globals.tem_host, 3535,  verbose=True)
+        self.client = TEMClient(globals.tem_host, globals.tem_port,  verbose=True)
         self.cfg = ConfigurationClient(redis_host(), token=auth_token())
-        self.metadata_notifier = MetadataNotifier(host = "noether", port = 3463, verbose = False)
+        self.metadata_notifier = MetadataNotifier(host = globals.dataserver_host, port = globals.dataserver_port, verbose = False)
 
         self.reset_rotation_signal.connect(self.tem_action.reset_rotation_button)
 
@@ -206,8 +206,8 @@ def run(self):
                                       beam_property,
                                       self.rotations_angles,
                                       self.cfg.threshold,
-                                      retries=3, 
-                                      delay=0.1) 
+                                      retries=globals.max_retries_tagging, 
+                                      delay=globals.inquiry_delay) 
 
                     self.file_operations.update_xtalinfo_signal.emit('Processing', 'XDS')
                     # self.file_operations.update_xtalinfo_signal.emit('Processing', 'DIALS')

jungfrau_gui/ui_components/tem_controls/task/stage_centering_task.py

@@ -19,16 +19,13 @@ def __init__(self, control_worker, pixels=[10, 1]):
         self.control = control_worker
         self.pixels = pixels
         logging.info("CenteringTask initialized")
-        self.client = TEMClient(globals.tem_host, 3535,  verbose=True)
+        self.client = TEMClient(globals.tem_host, globals.tem_port, verbose=True)
         self.cfg = ConfigurationClient(redis_host(), token=auth_token())
         for shape in self.cfg.overlays:
             if shape['type'] == 'rectangle':
                 self.lowmag_jump = shape['xy'][0]+shape['width']//2, shape['xy'][1]+shape['height']//2
                 break
-        self.thresholds = {
-            'dxy_min': 0.3, 'dxy_max': 100, 
-            'dz_min_mag': 1, 'dz_max_mag': 10, 'dz_min_lmag': 3, 'absz_min': -70, 'absz_max': 20, 
-        }
+        self.thresholds = globals.click_on_move_thresholds
 
     def rot2d(self, vector, theta):# anti-clockwise
         theta_r = np.radians(theta)
@@ -39,7 +36,7 @@ def rot2d(self, vector, theta):# anti-clockwise
     def translationvector(self, pixels, magnification):
         calibrated_mag = cfg_jf.lut().calibrated_magnification(magnification[2])
         rotation_axis = cfg_jf.lut().rotaxis_for_ht_degree(self.control.tem_status["ht.GetHtValue"], magnification=magnification[0])
-        if int(magnification[0]) >= 1500 : # Mag
+        if int(magnification[0]) >= globals.min_mag_for_mag: # Mag
             logging.debug(f'Estimate with rotation')
             tr_vector = (pixels - [self.cfg.ncols/2, self.cfg.nrows/2]) * globals.PIXEL * globals.MM_TO_UM / calibrated_mag # in um
         else: # Lowmag, targeting to the rectangular overlay

jungfrau_gui/ui_components/tem_controls/task/task_manager.py

@@ -63,7 +63,7 @@ class ControlWorker(QObject):
     def __init__(self, tem_action): #, timeout:int=10, buffer=1024):
         super().__init__()
         self.cfg = ConfigurationClient(redis_host(), token=auth_token())
-        self.client = TEMClient(globals.tem_host, 3535,  verbose=False)
+        self.client = TEMClient(globals.tem_host, globals.tem_port,  verbose=False)
 
         self.task = Task(self, "Dummy")
         self.task_thread = QThread()
@@ -421,7 +421,7 @@ def update_tem_status(self, response):
             position = np.array(pos_list)
             position_prev = np.array(pos_prev_list)
             diff_pos = position - position_prev
-            threshold = np.array([30, 30, 30, 0.2, 100])  # nm, nm, nm, deg., deg.
+            threshold = np.array(globals.stage_relaxation_thresholds)
             update_mask = np.abs(diff_pos) > threshold
 
             # Update diff using vectorized operations