import copy import time from math import ceil from typing import List, Tuple import cv2 import numpy as np import redis from aaredaq import workflows from aaredaq.autofocus import calculate_focus_measure from aaredaq.config import BeamlineConfig, ABR_POS_MOUNT from aaredaq.config import BeamlineStateEnum from aaredaq.devices import BeamlineDevices from aaredaq.mlbox import MlBox from aaredaqlib.beamline import MXBeamline from aaredaqlib.coordinate import Coordinate, SmargonCoordinate from aaredaqlib.diffraction_geometry import DiffractionGeometry from aaredaqlib.models import ( SampleShortInfo, PuckLoadedInfo, SampleShortInfoList, DAQStatusModel, BeamlineStatus, SessionStatus, SampleCameraSettings, AutofocusSettings, ) from aaredaqlib.raster_grid import RasterGridRequest, CompletedRasterGrid from aaredaqlib.rotation_scan import RotationScanRequest from aaredaqlib.sample_geometry import SampleGeometryModel from mxlibs3.jfjoch import JFJochWrapper class TransformationInvalidException(Exception): def __init__(self, message="Transformation is not implemented"): super().__init__(message) self.message = message def __str__(self): return self.message class LoopCenteringFailed(Exception): pass class AareDAQ: def __init__(self, cfg: BeamlineConfig, bl: MXBeamline): self.last_time = 0.0 self.__cfg = cfg self.__devs = BeamlineDevices(bl) self.__mlbox = MlBox() self.__jfjoch = JFJochWrapper(bl) self.__bl = bl.value.upper() @property def state(self) -> BeamlineStateEnum: return self.__cfg.state @property def busy(self) -> bool: return self.__cfg.state_busy @state.setter def state(self, target: BeamlineStateEnum): if target == BeamlineStateEnum.Moving: raise Exception("Cannot explicitly move to busy state") start = time.perf_counter() self.__cfg.try_set_busy(timeout=300) self.__set_state(target) self.__cfg.state_busy = False end = time.perf_counter() self.last_time = end - start @property def omega(self) -> float: return self.__devs.aerotech.omega @omega.setter def omega(self, val: float): self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment) if -720 < val < 720: self.__devs.aerotech.move(val, wait=True, speed=180.0, direct=True) self.__cfg.state_busy = False else: self.__cfg.state_busy = False raise ValueError("Omega has to be between -720 and 720 degrees (for now)") @property def zoom(self) -> float: return self.__devs.zoom @zoom.setter def zoom(self, val: float): self.__devs.zoom = val @property def light(self) -> float: val = self.__devs.lamp_light.value if val <= 1.0: return 0 elif val >= 2.5: return 100.0 else: return (val - 1.0) / 1.5 * 100.0 @light.setter def light(self, f: float): conv = (f / 100.0 * 1.5) + 1.0 print(f"Light {f} -> {conv}") self.__devs.lamp_light.move(conv, wait=True) @property def sample(self) -> SampleShortInfo | None: if self.__devs.tell.get_mounted_sample() is None: return None return self.__cfg.current_sample @property def samcam_settings(self) -> SampleCameraSettings: return self.__devs.samcam_settings @samcam_settings.setter def samcam_settings(self, s: SampleCameraSettings): self.__devs.samcam_settings = s def autofocus(self, f: AutofocusSettings): pos = self.__cfg.abr_meas_pos z_min = pos.z - f.z_range_um / 2.0 z_max = pos.z + f.z_range_um / 2.0 measures = [] for i in range(f.z_steps): pos.z = z_min + i * (z_max - z_min) / f.z_steps self.__devs.abr_pos = pos self.__devs.sample_cam.get_single_image() image = self.__devs.sample_cam.get_image(gray=True) val = calculate_focus_measure(image, f.center_x_pxl, f.center_y_pxl, f.radius_pxl) measures.append((pos.z, val)) best_pos, _ = max(measures, key=lambda x: x[1]) pos.z = best_pos self.__devs.abr_pos = pos self.__cfg.abr_meas_pos = pos self.__devs.sample_cam.collect_auto() def tweak_abr_meas_pos(self, c: Coordinate): self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment) try: new_meas_pos = self.__cfg.abr_meas_pos + c self.__cfg.abr_meas_pos = new_meas_pos self.__devs.abr_pos = new_meas_pos self.__cfg.state_busy = False except: self.__cfg.state_busy = False raise def save_abr_meas_pos(self): self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment) try: self.__cfg.abr_meas_pos = self.__devs.abr_pos self.__cfg.state_busy = False except: self.__cfg.state_busy = False raise def goto_abr_meas_pos(self): self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment) try: self.__devs.abr_pos = self.__cfg.abr_meas_pos self.__cfg.state_busy = False except: self.__cfg.state_busy = False raise def __mount(self, target: SampleShortInfo | None): self.__set_state(BeamlineStateEnum.RobotSampleExchange) self.__devs.smargon.move_home(wait=True) self.__devs.abr_pos = ABR_POS_MOUNT time.sleep(0.5) print(f"Smargon x={self.__devs.smargon.readback.sh_mm.x}, y={self.__devs.smargon.readback.sh_mm.y}, z={self.__devs.smargon.readback.sh_mm.z}") print(f"ABR x={self.__devs.abr_pos.x}, y={self.__devs.abr_pos.y}, z={self.__devs.abr_pos.z} omega={self.__devs.aerotech.omega}") if self.__devs.bsz.value < 24.0: raise Exception("Beamstop Z below 24.0 mm - potentially unsafe with mounting") if self.__devs.magnet_position_sensor.value != 0: raise Exception("Magnet position sensor is not in position") self.__devs.tell.check_enable_motion() self.__devs.tell.set_in_mount_position(True) curr_sample = self.__cfg.current_sample self.__devs.tell.wait_ready() if target is None: self.__devs.tell.unmount(wait=True, timeout=360) if curr_sample is not None: self.__devs.aare.sample_unmounted(curr_sample) self.__cfg.current_sample = None else: if target.location is None: raise Exception("Sample not loaded into dewar") if curr_sample is not None: self.__devs.aare.sample_unmounted(curr_sample) self.__devs.tell.mount( address=target.tell_address(), force=True, auto_unmount=True, read_dm=False, wait=True, timeout=360, ) self.__devs.aare.sample_mounted(target) self.__cfg.current_sample = target @sample.setter def sample(self, target: SampleShortInfo | None): # This will set internally state to SampleExchange, but will return to SampleAlignment # before exiting start = time.perf_counter() self.__cfg.try_set_busy(timeout=360) try: self.__mount(target) self.__cfg.state_busy = False except Exception as e: self.__cfg.state_busy = False # If unmount succeeded, but mount failed if self.__devs.tell.get_mounted_sample() is None: self.__cfg.current_sample = None raise e self.__set_state(BeamlineStateEnum.SampleAlignment) end = time.perf_counter() self.last_time = end - start @property def camera_image(self) -> np.ndarray | None: image = self.__devs.sample_cam.get_image(gray=False) return image @property def camera_image_gray(self) -> np.ndarray | None: image = self.__devs.sample_cam.get_image(gray=True) return image def list_loaded_pucks(self) -> List[PuckLoadedInfo]: return self.__devs.tell.get_detected_pucks() def __raster(self, r: RasterGridRequest) -> CompletedRasterGrid: max_time = r.exp_time_s * r.n_y * r.n_x + 60 self.__cfg.dtz = r.dtz self.__set_state(BeamlineStateEnum.DataCollection) self.__jfjoch.measure_raster(r, self.diffraction_geometry, self.sample) self.__devs.aerotech.move(r.omega_deg, wait=True, speed=360.0) save_smargon_position = self.__devs.smargon.readback self.__devs.smargon.target = r.smargon self.__devs.aerotech.measure_raster_simple( r.exp_time_s, r.grid_size_mm.x, r.grid_size_mm.y, r.n_x, r.n_y ) self.__devs.aerotech.wait_scan_done(max_time) self.__devs.aerotech.reset() self.__devs.abr_pos = self.__cfg.abr_meas_pos self.__devs.smargon.target = save_smargon_position result = self.__jfjoch.wait_till_done(5) return CompletedRasterGrid(request=copy.deepcopy(r), result=result) def measure_raster(self, r: RasterGridRequest) -> CompletedRasterGrid: max_time = r.exp_time_s * r.n_y * r.n_x + 60 self.__cfg.try_set_busy(timeout=ceil(max_time + 360)) try: result = self.__raster(r) self.__set_state(BeamlineStateEnum.SampleAlignment) self.__cfg.state_busy = False return result except Exception as e: self.__set_state(BeamlineStateEnum.SampleAlignment) self.__cfg.state_busy = False raise e def __rotation(self, request: RotationScanRequest): self.__cfg.dtz = request.dtz self.__set_state(BeamlineStateEnum.DataCollection) if request.start is not None: self.__devs.smargon.target = request.start self.__devs.smargon.wait() self.__jfjoch.measure_rotation(request, self.diffraction_geometry, self.sample) if request.screening: self.__devs.aerotech.measure_screening(request.start_omega_deg, request.wedge_omega_deg, request.exp_time_s, request.incr_omega_deg, request.steps) self.__devs.aerotech.wait_scan_done(request.exp_time_s * request.steps + 60) else: total_omega = request.incr_omega_deg * request.steps total_time = request.exp_time_s * request.steps self.__devs.aerotech.measure_standard( request.start_omega_deg, total_omega, total_time ) if request.start is not None and request.end is not None: smargon_time_step = request.time_sec / float(request.steps) pos_step = (request.end.sh_mm - request.start.sh_mm) * (1.0 / float(request.steps)) for i in range(request.steps): self.__devs.smargon.target = SmargonCoordinate( sh_mm=request.start.sh_mm + pos_step * i ) time.sleep(smargon_time_step) self.__devs.aerotech.wait_scan_done(total_time + 60) self.__jfjoch.wait_till_done(60) self.__devs.aerotech.reset() def measure_rotation(self, request: RotationScanRequest): total_time = request.exp_time_s * request.steps self.__cfg.try_set_busy(timeout=ceil(total_time + 360)) try: self.__rotation(request) self.__set_state(BeamlineStateEnum.SampleAlignment) self.__cfg.state_busy = False except Exception as e: self.__set_state(BeamlineStateEnum.SampleAlignment) self.__cfg.state_busy = False raise e def get_background(self): #if self.sample is not None: # raise Exception("Background cannot be measured when sample is mounted") self.__cfg.try_set_busy(timeout=360) self.__set_state(BeamlineStateEnum.SampleAlignment) try: self.__devs.lamp_light.move("on", wait=True) for s in self.__cfg.alc_zoom_settings.z: self.__devs.samcam_settings = SampleCameraSettings(exposure=s.sam_cam_exp, gain=s.sam_cam_gain) self.__devs.zoom = s.zoom_value time.sleep(5.0) # Wait for settings to stabilize image = self.__devs.sample_cam.get_image(gray=False) self.__cfg.put_alc_bkg(s.zoom_value, image) bgr_array = cv2.cvtColor(image, cv2.COLOR_RGB2BGR) cv2.imwrite(f"bkg{s.zoom_value:.0f}.jpg", bgr_array) self.__cfg.state_busy = False self.__devs.sample_cam.set_auto() except Exception as e: self.__cfg.state_busy = False raise e @property def dtz(self) -> float: tmp = self.__cfg.dtz if tmp is None: return 150.0 else: return tmp @dtz.setter def dtz(self, val: float): self.__cfg.try_set_busy(timeout=360) state = self.__cfg.state if state == BeamlineStateEnum.DataCollection: self.__cfg.state_busy = False raise RuntimeError("Cannot set dtz during data collection") elif state == BeamlineStateEnum.SampleAlignment: self.__devs.dtz = val self.__cfg.dtz = val self.__cfg.state_busy = False @property def smargon(self) -> SmargonCoordinate: return self.__devs.smargon.readback @smargon.setter def smargon(self, sc: SmargonCoordinate): self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment) try: self.__devs.smargon.target = sc self.__devs.smargon.wait() self.__cfg.state_busy = False except Exception as e: self.__cfg.state_busy = False raise e pass def mark_beam(self, x_pxl: float, y_pxl: float): self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment) try: self.__cfg.mark_beam(x_pxl, y_pxl, self.__devs.zoom) self.__cfg.state_busy = False except Exception as e: self.__cfg.state_busy = False raise e def clear_mark_beam(self): self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment) self.__cfg.clear_mark_beam() self.__cfg.state_busy = False @property def sample_geometry(self) -> SampleGeometryModel: zoom = self.__devs.zoom abr_pos = self.__cfg.abr_meas_pos return SampleGeometryModel( beam_location_pxl=self.__cfg.beam_mark_coeff.apply(zoom), pixel_in_mm=self.__cfg.pixel_to_mm(zoom), omega_deg=self.__devs.aerotech.omega, smargon=self.__devs.smargon.readback, beam_size_mm=Coordinate(x=0.040, y=0.040), aerotech=Coordinate( x=self.__devs.gmx.value - abr_pos.x, y=self.__devs.gmy.value - abr_pos.y, z=self.__devs.gmz.value - abr_pos.z, ), aerotech_meas=self.__devs.abr_pos ) @property def beam_center(self) -> Tuple[float, float]: return self.__cfg.beam_center @beam_center.setter def beam_center(self, val: Tuple[float, float]): self.__cfg.beam_center = val def get_beam_mark(self): return self.__cfg.get_beam_mark(self.__devs.zoom) def listen_changes(self) -> redis.client.PubSub: return self.__cfg.listen_changes() def __ml_bounding_box(self, sample_id: int | None = None, filename: str | None = None, move: bool = False) -> None | Tuple[float, float, float, float]: time.sleep(0.2) # Just to be sure image is stable #curr_image = self.camera_image #box = self.__mlbox.predict(curr_image) bgr_image = cv2.cvtColor(self.camera_image, cv2.COLOR_RGB2BGR) box = self.__mlbox.predict(bgr_image) if box is None: if filename is not None: cv2.imwrite(f"{filename}_no_detection.jpg", bgr_image) return None x1, y1, x2, y2 = box if filename is not None: cv2.rectangle(bgr_image, (int(x1), int(y1)), (int(x2), int(y2)), (0, 255, 0), 2) cv2.imwrite(f"{filename}.jpg", bgr_image) if sample_id is not None: self.__devs.aare.upload_image(sample_id, filename, bgr_image) if move: geom = self.sample_geometry coord = geom.picture_to_smargon(Coordinate(x=x1, y=y1)) self.__devs.smargon.target = SmargonCoordinate(sh_mm=coord) self.__devs.smargon.wait(60) return box def ml_bounding_box(self, sample_id: int | None = None, filename: str | None = None, move: bool = False) -> None | Tuple[float, float, float, float]: try: self.__cfg.try_set_busy(timeout=360) box = self.__ml_bounding_box(sample_id, filename, move) if not box: raise LoopCenteringFailed self.__cfg.state_busy = False return box except Exception: self.__cfg.state_busy = False raise def __loop_center(self, filename: str | None = "") -> SmargonCoordinate: curr_image = self.camera_image gray_without_feature = cv2.cvtColor( self.__cfg.get_alc_bkg(self.zoom), cv2.COLOR_RGB2GRAY ) gray_with_feature = cv2.cvtColor(curr_image, cv2.COLOR_RGB2GRAY) # Subtract the "without feature" image from the "with feature" image diff_image = cv2.absdiff(gray_with_feature, gray_without_feature) # Threshold the difference to isolate the feature _, thresh = cv2.threshold(diff_image, 70, 255, cv2.THRESH_BINARY) # thresh = cv2.adaptiveThreshold(diff_image, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 11, 2) if filename is not None: cv2.imwrite(f"{filename}_diff.jpg", diff_image) cv2.imwrite(f"{filename}_thresh.jpg", thresh) # Find contours of the detected feature contours, _ = cv2.findContours( thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE ) leftmost_point = None # Loop through each contour for contour in contours: # Loop through each point in the current contour for point in contour: x, y = point[0] # Point is a nested array [ [x, y] ] # Check if this is the leftmost point if leftmost_point is None or x < leftmost_point[0]: leftmost_point = (x, y) if leftmost_point is None: raise LoopCenteringFailed() geom = self.sample_geometry coord = geom.picture_to_smargon( Coordinate(x=leftmost_point[0], y=leftmost_point[1]) ) return SmargonCoordinate(sh_mm=coord) def __loop_center_sequence(self, sample_id: int | None = None) -> bool: self.__set_state(BeamlineStateEnum.SampleAlignment) self.__devs.lamp_light.move("on", wait=True) try: i = 0 for s in self.__cfg.alc_zoom_settings.z: self.__devs.samcam_settings = SampleCameraSettings(exposure=s.sam_cam_exp, gain=s.sam_cam_gain) self.__devs.zoom_sync(s.zoom_value) if i % 2 == 0: angles = (0, 90) else: angles = (90, 0) for angle in angles: self.__devs.aerotech.move(angle, wait=True) filename = None if sample_id is not None: filename = f"{angle}_{s.zoom_value:.0f}" self.__devs.smargon.target = self.__loop_center(filename) self.__devs.smargon.wait(60) if sample_id is not None: time.sleep(0.1) self.save_screenshot_db(sample_id, f"{angle}_{s.zoom_value:.0f}") i += 1 return True except LoopCenteringFailed: return False def auto_loop_center(self, sample_id: int | None = None) -> float: start = time.perf_counter() try: self.__cfg.try_set_busy(timeout=360) if not self.__loop_center_sequence(sample_id): raise LoopCenteringFailed self.__cfg.state_busy = False except Exception: self.__cfg.state_busy = False raise end = time.perf_counter() return end - start def save_screenshot(self, filename: str): time.sleep(0.2) # Wait 200 ms to ensure camera image is stable bgr_image = cv2.cvtColor(self.camera_image, cv2.COLOR_RGB2BGR) cv2.imwrite(f"{filename}.jpg", bgr_image) def save_screenshot_db(self, sample_id: int, filename: str): time.sleep(0.2) # Wait 200 ms to ensure camera image is stable bgr_image = cv2.cvtColor(self.camera_image, cv2.COLOR_RGB2BGR) self.__devs.aare.upload_image(sample_id, filename, bgr_image) @property def sample_spreadsheet(self) -> SampleShortInfoList: return SampleShortInfoList(s=[]) #return self.__devs.aare.get_sample_info() def sample_spreadsheet_user(self, pgroup: str) -> SampleShortInfoList: sample = self.sample_spreadsheet sample.s = list(filter(lambda x: x.user == pgroup, sample.s)) return sample def measure(self, sample: SampleShortInfo) -> float: start = time.perf_counter() if sample.location is not None: sample_name = "{}{}S{:02d}_{}".format( sample.location.segment, sample.location.pos, sample.pin, sample.sample_name ) else: sample_name = sample.sample_name try: self.__cfg.try_set_busy(timeout=360) self.__mount(sample) self.__set_state(BeamlineStateEnum.SampleAlignment) self.save_screenshot_db(sample.db_id, "mount.jpeg") self.__loop_center_sequence(sample.db_id) self.save_screenshot_db(sample.db_id, "after_lc.jpeg") self.__set_state(BeamlineStateEnum.DataCollection) self.__devs.aerotech.move(0, wait=True) self.__ml_bounding_box(sample.db_id, f"bb_raster_0", move=True) self.__raster( RasterGridRequest( n_x=25, n_y=15, grid_size_mm=Coordinate(x=0.02, y=0.02), smargon=SmargonCoordinate(), omega_deg=0, exp_time_s=0.01, ) ) self.__devs.aerotech.move(90, wait=True) self.__ml_bounding_box(sample.db_id, f"bb_raster_90", move=True) self.__raster( RasterGridRequest( n_x=25, n_y=10, grid_size_mm=Coordinate(x=0.02, y=0.02), smargon=SmargonCoordinate(), omega_deg=90, exp_time_s=0.01, ) ) #self.__rotation( # RotationScanRequest(total_omega_deg=360, # start_omega_deg=0, # total_time_sec=18)) self.save_screenshot_db(sample.db_id, "after_dc.jpeg") self.__set_state(BeamlineStateEnum.RobotSampleExchange) self.__cfg.state_busy = False except Exception: self.__cfg.state_busy = False raise end = time.perf_counter() return end - start def __set_state(self, target: BeamlineStateEnum): # __set_state assumes that beamline is already in busy state # it will apply a proper transformation and change state afterward # specifically: # 1. If target state is maintenance, just go there # 2. If target state is same as current, nothing will happen # 3. If target state cannot be reached, exception is raised and current state is kept # 4. If exception is raised during transformation, state is set to maintenance # 5. If transformation goes OK, target state is set # # Busy state will be cleared only, if exception is raised. # curr_state = self.__cfg.state if target == BeamlineStateEnum.Maintenance: self.__cfg.state = BeamlineStateEnum.Maintenance elif target != curr_state: self.__cfg.state = BeamlineStateEnum.Moving try: match curr_state: case BeamlineStateEnum.Maintenance: if target == BeamlineStateEnum.SampleExchange: workflows.m2se(self.__devs, self.__cfg) else: raise TransformationInvalidException() case BeamlineStateEnum.SampleExchange: if target == BeamlineStateEnum.SampleAlignment: workflows.se2sa(self.__devs, self.__cfg) elif target == BeamlineStateEnum.BeamLocation: workflows.se2sa(self.__devs, self.__cfg) workflows.sa2bl(self.__devs, self.__cfg) else: raise TransformationInvalidException() case BeamlineStateEnum.DewarTransfer: if target == BeamlineStateEnum.SampleAlignment: workflows.dh2sa(self.__devs, self.__cfg) else: raise TransformationInvalidException() case BeamlineStateEnum.DataCollection: if target == BeamlineStateEnum.SampleExchange: workflows.dc2se(self.__devs, self.__cfg) elif target == BeamlineStateEnum.SampleAlignment: workflows.dc2sa(self.__devs, self.__cfg) elif target == BeamlineStateEnum.RobotSampleExchange: workflows.dc2rse(self.__devs, self.__cfg) else: raise TransformationInvalidException() case BeamlineStateEnum.BeamLocation: if target == BeamlineStateEnum.SampleAlignment: workflows.bl2sa(self.__devs, self.__cfg) if target == BeamlineStateEnum.SampleExchange: workflows.bl2sa(self.__devs, self.__cfg) workflows.sa2se(self.__devs, self.__cfg) else: raise TransformationInvalidException() case BeamlineStateEnum.SampleAlignment: if target == BeamlineStateEnum.DewarTransfer: workflows.sa2dh(self.__devs, self.__cfg) elif target == BeamlineStateEnum.SampleExchange: workflows.sa2se(self.__devs, self.__cfg) elif target == BeamlineStateEnum.RobotSampleExchange: workflows.sa2rse(self.__devs, self.__cfg) elif target == BeamlineStateEnum.DataCollection: workflows.sa2dc(self.__devs, self.__cfg) elif target == BeamlineStateEnum.XrayFluorescence: workflows.sa2xrf(self.__devs, self.__cfg) elif target == BeamlineStateEnum.BeamLocation: workflows.sa2bl(self.__devs, self.__cfg) else: raise TransformationInvalidException() case BeamlineStateEnum.XrayFluorescence: if target == BeamlineStateEnum.SampleAlignment: workflows.xrf2sa(self.__devs, self.__cfg) else: raise TransformationInvalidException() case BeamlineStateEnum.RobotSampleExchange: if target == BeamlineStateEnum.SampleAlignment: workflows.se2sa(self.__devs, self.__cfg) else: raise TransformationInvalidException() self.__cfg.state = target except TransformationInvalidException as e: self.__cfg.state = curr_state self.__cfg.state_busy = False raise e except Exception as e: self.__cfg.state = BeamlineStateEnum.Maintenance self.__cfg.state_busy = False raise e def close_shutter(self): self.__devs.shutter = False @property def diffraction_geometry(self) -> DiffractionGeometry: det_cfg = self.__jfjoch.detector() return DiffractionGeometry( energy_keV=self.__devs.energy_kev, dtz_mm=self.__devs.dtz.value, detector_size_pxl=(det_cfg.width, det_cfg.height), pixel_size_mm=det_cfg.pixel_size_mm, beam_center_pxl=self.__cfg.beam_center, detector_description=det_cfg.description, detector_serial_number=det_cfg.serial_number ) @property def beamline_status(self) -> BeamlineStatus: return BeamlineStatus( ring_current_mA=self.__devs.ring_current, light=self.light, cryojet_K=self.__devs.cryojet_temp, shutter_open=self.__devs.shutter, flux_ph_s=3.0e12, sample_camera=self.__devs.samcam_settings, name=self.__bl ) @property def status(self) -> DAQStatusModel: # session should be set by FastAPI server return DAQStatusModel( state=self.state, busy=self.busy, geom=self.sample_geometry, bl=self.beamline_status, sample=self.sample, session=SessionStatus(), diffraction=self.diffraction_geometry ) def cancel(self): if self.__cfg.state == BeamlineStateEnum.DataCollection: self.__devs.aerotech.stop() self.__jfjoch.cancel()