mx/AareDAQ
0
1
Fork 0
Code Issues 49 Pull Requests Actions Packages Projects Releases 84 Wiki Activity

jfjoch: refactored so fomrat_dataset_settings is one funcution, added newenum for scantypes.

Browse Source
This commit is contained in:
appleb_m
2026-04-28 14:14:18 +02:00
parent 386df67557
commit 98fba7ec53
1 changed files with 151 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aare/devices/jfjoch.py
+151 -52
View File
@@ -1,5 +1,6 @@
import math
import time
from enum import Enum
import jfjoch_client
@@ -9,6 +10,13 @@ from aare.common.models import DAQStatusModel, FluorescenceSpectrumOutputModel
from aare.common.raster_grid import RasterGridRequest
from aare.common.rotation_scan import RotationScanRequest
class ScanTypeEnum(Enum):
RASTER = "Raster"
XRF = "XRF"
ROTATION = "Rotation"
SCREENING = "Screening"
HELICAL = "Helical"
UNKNOWN = "Unknown"
class JFJochWrapper:
def __init__(self, bl: MXBeamline):
@@ -40,10 +48,14 @@ class JFJochWrapper:
status = self.__api.status_get()
return status.state == 'Idle'
def measure_rotation(self,
r: RotationScanRequest,
s: DAQStatusModel,
f: FluorescenceSpectrumOutputModel | None = None) -> None:
def __format_dataset_settings(
self,
r: RasterGridRequest | RotationScanRequest,
s: DAQStatusModel,
f: FluorescenceSpectrumOutputModel | None = None,
async_start: bool = True
) -> jfjoch_client.DatasetSettings:
#common sample parameter intiialisation
if s.sample is None:
pgroup = "p16371"
sample = "unknown_sample"
@@ -51,84 +63,113 @@ class JFJochWrapper:
pgroup = s.sample.user
sample = s.sample.sample_name
if r.screening:
wedge = r.wedge_omega_deg
#raster grid, rotation or screening specific parameter initialisation
wedge = None
if isinstance(r, RasterGridRequest):
if r.n_x == 1:
trigger = 1
images = r.n_y
else:
trigger = r.n_y
images = r.n_x
else:
wedge = None
goniometer_settings = jfjoch_client.RotationAxis(
step=r.incr_omega_deg,
start=r.start_omega_deg,
name="omega",
vector= [-1,0,0],
screening_wedge_deg = wedge
)
if r.screening:
images = 1
trigger = r.steps
else:
images = r.steps
trigger = 1
xrf = None
if f is not None:
xrf = jfjoch_client.DatasetSettingsXrayFluorescenceSpectrum(
energy_eV=f.energy_eV,
data=f.spectrum
)
if r.screening:
wedge = r.wedge_omega_deg
images = 1
trigger = r.steps
else:
wedge = None
images = r.steps
trigger = 1
#build common dataset settings
dataset_settings = jfjoch_client.DatasetSettings(
beam_x_pxl=s.diffraction.beam_center_pxl[0],
beam_y_pxl=s.diffraction.beam_center_pxl[1],
ntrigger= trigger,
images_per_trigger=images,
detector_distance_mm=s.diffraction.dtz_mm,
file_prefix=f"{pgroup}/raw/{r.file_prefix}",
file_prefix=f"{pgroup}/raw/raster/{r.file_prefix}",
incident_energy_keV=s.diffraction.energy_keV,
goniometer=goniometer_settings,
sample_name=sample,
image_time_us=round(r.exp_time_s * 1e6),
experiment_group=pgroup,
transmission=s.bl.transmission,
ring_current_mA=s.bl.ring_current_mA,
sample_temperature_K=s.bl.cryojet_K,
total_flux=s.bl.flux_ph_s,
poni_rot1_rad=s.diffraction.poni_rot1_rad,
poni_rot2_rad=s.diffraction.poni_rot2_rad,
total_flux=s.bl.flux_ph_s,
max_spot_count=1000,
detect_ice_rings=True,
xray_fluorescence_spectrum=xrf
async_start=async_start
)
#settup grid or goniometer settings adn add to dataset settings
if isinstance(r, RasterGridRequest):
grid_settings = jfjoch_client.GridScan(
n_fast=r.n_x,
snake=True,
step_x_um=r.grid_size_mm.x * 1000.0,
step_y_um=r.grid_size_mm.y * 1000.0,
vertical=False
)
dataset_settings.grid_scan = grid_settings
else:
goniometer_settings = jfjoch_client.RotationAxis(
step=r.incr_omega_deg,
start=r.start_omega_deg,
name="omega",
vector=[-1, 0, 0],
screening_wedge_deg=wedge
)
dataset_settings.goniometer = goniometer_settings
#if measuring fluoresence, add to dataset settings
if f is not None:
xrf = jfjoch_client.DatasetSettingsXrayFluorescenceSpectrum(
energy_eV=f.energy_eV,
data=f.spectrum
)
dataset_settings.xray_fluorescence_spectrum = xrf
return dataset_settings
def __start_scan(
self,
scan_type: ScanTypeEnum,
r: RasterGridRequest | RotationScanRequest,
s: DAQStatusModel,
f: FluorescenceSpectrumOutputModel | None = None,
async_start: bool = True
):
dataset_settings = self.__format_dataset_settings(r, s, f, async_start=async_start)
try:
self.__api.start_post(dataset_settings=dataset_settings)
except Exception as e:
scan_type = "rotation"
if r.screening:
scan_type = "screening"
raise JFJochCommunicationError(
f"JFJoch data collection failed to initialize for {scan_type} scan",
f"JFJoch data collection failed to initialize for {scan_type.value} scan with exception: {e}",
operation="POST",
endpoint="start_post",
base_url=self.__url,
) from e
def measure_rotation(self,
r: RotationScanRequest,
s: DAQStatusModel,
f: FluorescenceSpectrumOutputModel | None = None,
async_start:bool = True) -> None:
if r.screening:
self.__start_scan(ScanTypeEnum.SCREENING, r, s, f, async_start=async_start)
else:
self.__start_scan(ScanTypeEnum.ROTATION, r, s, f, async_start=async_start)
def measure_raster(self,
r: RasterGridRequest,
s: DAQStatusModel):
if s.sample is None:
pgroup = "p16371"
sample = "unknown_sample"
else:
pgroup = s.sample.user
sample = s.sample.sample_name
if r.n_x == 1:
trigger = 1
images = r.n_y
else:
trigger = r.n_y
images = r.n_x
s: DAQStatusModel,
async_start: bool = True):
self.__start_scan(ScanTypeEnum.RASTER, r, s, async_start=async_start)
def wait_till_running(self, timeout: int | float = 60):
if self.__simulated:
@@ -174,4 +215,62 @@ class JFJochWrapper:
except Exception as e:
last_error = e
time.sleep(wait_between_retries_s)
raise last_error
raise last_error
if __name__ == "__main__":
import logging
from aare.common.coordinate import SmargonCoordinate, Coordinate
from aare.common.models import SessionStatus, BeamlineStateEnum, SampleCameraSettings, BeamlineStatus
from aare.common.sample_geometry import SampleGeometryModel
from aare.common.diffraction_geometry import DiffractionGeometry
logging.basicConfig(level=logging.DEBUG)
bl = MXBeamline.X10SA
wrapper = JFJochWrapper(bl)
#wrapper.initialize()
#wrapper.wait_till_done(timeout=360)
status = DAQStatusModel(
bl=BeamlineStatus(
name="X10SA",
ring_current_mA=400.0,
flux_ph_s=1e12,
transmission=1.0,
cryojet_K=100.0,
front_light=50.0,
back_light=50.0,
shutter_open=False,
exp_shutter_open=False,
sample_camera=SampleCameraSettings(gain=1.0, exposure=0.02),
zoom=1.0,
commissioning_mode=False,
dtz_min=120.0,
dtz_max=1600.0
),
diffraction=DiffractionGeometry(
detector_description="EIGER",
detector_serial_number="123",
dtz_mm=200.0,
pixel_size_mm=0.075,
energy_keV=12.658,
beam_center_pxl=(1000.0, 1000.0),
detector_size_pxl=(4000, 4000),
poni_rot1_rad=0.0,
poni_rot2_rad=0.0
),
geom=SampleGeometryModel(
beam_location_pxl=Coordinate(x=1000, y=1000),
pixel_in_mm=0.001,
aerotech=Coordinate(x=0, y=0),
aerotech_meas=Coordinate(x=0, y=0),
smargon=SmargonCoordinate(sh_mm=Coordinate(x=0,y=0,z=0), phi_deg=0, chi_deg=0),
omega_deg=10.5,
beam_size_mm=Coordinate(x=0.02, y=0.01)
),
session=SessionStatus(),
state=BeamlineStateEnum.Maintenance,
busy=False
)
wrapper.measure_raster(RasterGridRequest(n_x=1, n_y=1, grid_size_mm=Coordinate(x=0.05,y=0.05), file_prefix="test", exp_time_s=0.1,dtz=150,transmission=1,smargon_top_left=SmargonCoordinate(),omega_deg=0.0), status)
print("waiting for detector to start")
wrapper.wait_till_running()
print("success we can measure")
wrapper.cancel()