110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
import math
|
|
|
|
import jfjoch_client
|
|
|
|
from aaredaqlib.beamline import MXBeamline
|
|
from aaredaqlib.diffraction_geometry import DiffractionGeometry
|
|
from aaredaqlib.models import SampleShortInfo
|
|
from aaredaqlib.raster_grid import RasterGridRequest
|
|
from aaredaqlib.rotation_scan import RotationScanRequest
|
|
|
|
|
|
class JFJochWrapper:
|
|
def __init__(self, bl: MXBeamline):
|
|
match bl:
|
|
case MXBeamline.X06DA:
|
|
self.__url = "http://sls-gpu-001:8080"
|
|
case MXBeamline.SIMULATED:
|
|
self.__url = "http://localhost:8080"
|
|
case _:
|
|
raise Exception("unknown beamline")
|
|
|
|
self.__client = jfjoch_client.ApiClient(jfjoch_client.Configuration(host=self.__url))
|
|
self.__api = jfjoch_client.DefaultApi(self.__client)
|
|
|
|
def initialize(self):
|
|
self.__api.initialize_post()
|
|
|
|
def is_idle(self) -> bool:
|
|
status = self.__api.status_get()
|
|
return status.state == 'Idle'
|
|
|
|
def measure_rotation(self,
|
|
r: RotationScanRequest,
|
|
d: DiffractionGeometry,
|
|
s: SampleShortInfo | None) -> None:
|
|
if s is None:
|
|
pgroup = "p16371"
|
|
sample = None
|
|
else:
|
|
pgroup = s.user
|
|
sample = s.sample_name
|
|
|
|
goniometer_settings = jfjoch_client.RotationAxis(
|
|
step=r.incr_omega_deg,
|
|
start=r.start_omega_deg,
|
|
name="omega",
|
|
vector= [0,1,0]
|
|
)
|
|
|
|
dataset_settings = jfjoch_client.DatasetSettings(
|
|
beam_x_pxl=d.beam_center_pxl[0],
|
|
beam_y_pxl=d.beam_center_pxl[1],
|
|
ntrigger= 1,
|
|
images_per_trigger=r.steps,
|
|
detector_distance_mm=d.dtz_mm,
|
|
file_prefix=f"{pgroup}/raw/{r.file_prefix}",
|
|
incident_energy_keV=d.energy_keV,
|
|
goniometer=goniometer_settings,
|
|
sample_name=sample,
|
|
image_time_us=round(r.exp_time_s * 1e6),
|
|
experiment_group=pgroup
|
|
)
|
|
self.__api.start_post(dataset_settings=dataset_settings)
|
|
|
|
def measure_raster(self,
|
|
r: RasterGridRequest,
|
|
d: DiffractionGeometry,
|
|
s: SampleShortInfo | None):
|
|
if s is None:
|
|
pgroup = "p16371"
|
|
sample = None
|
|
else:
|
|
pgroup = s.user
|
|
sample = s.sample_name
|
|
|
|
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 = jfjoch_client.DatasetSettings(
|
|
beam_x_pxl=d.beam_center_pxl[0],
|
|
beam_y_pxl=d.beam_center_pxl[1],
|
|
ntrigger= r.n_y,
|
|
images_per_trigger=r.n_x,
|
|
detector_distance_mm=d.dtz_mm,
|
|
file_prefix=f"{pgroup}/raw/{r.file_prefix}",
|
|
incident_energy_keV=d.energy_keV,
|
|
grid_scan=grid_settings,
|
|
sample_name=sample,
|
|
image_time_us=round(r.exp_time_s * 1e6),
|
|
experiment_group=pgroup
|
|
)
|
|
self.__api.start_post(dataset_settings=dataset_settings)
|
|
|
|
def wait_till_done(self, timeout : int | float) -> jfjoch_client.models.ScanResult:
|
|
self.__api.wait_till_done_post_with_http_info(timeout=math.ceil(timeout))
|
|
return self.__api.result_scan_get()
|
|
|
|
def detector(self) -> jfjoch_client.models.DetectorListElement:
|
|
l = self.__api.config_select_detector_get()
|
|
|
|
if len(l.detectors) == 0:
|
|
raise Exception("no detectors configured")
|
|
|
|
return l.detectors[l.current_id]
|