Added functions to create rotation and gridscans runs into aaredb.py
Build and Publish / build (push) Successful in 22s

This commit is contained in:
GotthardG
2025-06-18 13:23:19 +02:00
parent c670668dd0
commit d66f758ff3
+75 -25
View File
@@ -11,8 +11,10 @@ from aareDBclient import (
SetTellPosition,
SampleEventCreate,
SetTellPositionRequest,
CharacterizationParameters,
RotationParameters,
GridScanParameters,
Datasets,
Detector as DetectorParameters,
BeamlineParameters,
ExperimentParametersCreate)
@@ -148,6 +150,66 @@ class AareWrapper:
verify=False, headers=headers)
print(response.status_code)
def create_characterization_run(self, s: Optional[SampleShortInfo], r:RotationScanRequest, d:DAQStatusModel):
if s is None:
return
try:
characterization = CharacterizationParameters(
omegaStart_deg=r.start_omega_deg,
omegaStep=r.incr_omega_deg,
phi=d.geom.smargon.phi_deg,
chi=d.geom.smargon.chi_deg,
numberOfImages=r.steps,
exposureTime_s=r.exp_time_s,
)
dataset = Datasets(
filepath=r.file_prefix,
status="written",
written_at=datetime.datetime.now()
)
detector_data = DetectorParameters(
manufacturer="DECTRIS",
model=d.diffraction.detector_description,
type="photon-counting",
serialNumber=d.diffraction.detector_serial_number,
detectorDistance_mm=r.dtz,
beamCenterX_px=d.diffraction.beam_center_pxl[0],
beamCenterY_px=d.diffraction.beam_center_pxl[1],
pixelSizeX_um=d.diffraction.pixel_size_mm * 1000,
pixelSizeY_um=d.diffraction.pixel_size_mm * 1000,
dataset=dataset
)
beamline_params = BeamlineParameters(
synchrotron="Swiss Light Source",
beamline=d.bl.name,
detector=detector_data,
energy_keV=d.diffraction.energy_keV,
wavelength_Ang=d.diffraction.wavelength_angstrom,
ringCurrent_mA=d.bl.ring_current_mA,
ringMode="Beamline Development",
monochromator="Si111",
transmission=r.transmission,
focusingOptic="Kirkpatrick-Baez",
beamlineFluxAtSample_ph_s=d.bl.flux_ph_s,
beamSizeWidth=d.geom.beam_size_mm.x * 100,
beamSizeHeight=d.geom.beam_size_mm.y * 100,
cryojetTemperature_K=d.bl.cryojet_K,
characterization=characterization,
)
experiment_params_payload = ExperimentParametersCreate(
type="standard",
beamline_parameters=beamline_params,
sample_id=s.db_id
)
response = self.__sample_api.create_experiment_parameters_for_sample(
sample_id=s.db_id,
experiment_parameters_create=experiment_params_payload
)
print("Experiment parameters created:", response)
except Exception as e:
print(e)
def create_rotation_run(self, s: Optional[SampleShortInfo], r:RotationScanRequest, d:DAQStatusModel):
if s is None:
return
@@ -161,6 +223,11 @@ class AareWrapper:
numberOfImages=r.steps,
exposureTime_s=r.exp_time_s,
)
dataset = Datasets(
filepath=r.file_prefix,
status="written",
written_at=datetime.datetime.now()
)
detector_data = DetectorParameters(
manufacturer="DECTRIS",
model=d.diffraction.detector_description,
@@ -171,6 +238,7 @@ class AareWrapper:
beamCenterY_px=d.diffraction.beam_center_pxl[1],
pixelSizeX_um=d.diffraction.pixel_size_mm * 1000,
pixelSizeY_um=d.diffraction.pixel_size_mm * 1000,
dataset=dataset
)
beamline_params = BeamlineParameters(
synchrotron="Swiss Light Source",
@@ -218,6 +286,11 @@ class AareWrapper:
numberOfImages=r.n_x * r.n_y,
exposureTime_s=r.exp_time_s,
)
dataset = Datasets(
filepath=r.file_prefix,
status="written",
written_at=datetime.datetime.now()
)
detector_data = DetectorParameters(
manufacturer="DECTRIS",
model=d.diffraction.detector_description,
@@ -228,6 +301,7 @@ class AareWrapper:
beamCenterY_px=d.diffraction.beam_center_pxl[1],
pixelSizeX_um=d.diffraction.pixel_size_mm * 1000,
pixelSizeY_um=d.diffraction.pixel_size_mm * 1000,
dataset=dataset
)
beamline_params = BeamlineParameters(
synchrotron="Swiss Light Source",
@@ -257,28 +331,4 @@ class AareWrapper:
)
print("Experiment parameters created:", response)
except Exception as e:
print(e)
def confirm_image_written(self, s: Optional[SampleShortInfo], r:RotationScanRequest):
if s is None:
return
try:
response = self.__sample_api.update_dataset_for_experiment_run(
filepath=r.file_prefix, #need to check if this is correct
status="written",
written_at=datetime.datetime.now().isoformat(),
)
print("Experiment run updated:", response)
except Exception as e:
print(e)
#def send_gridscan_result(self, s: Optional[SampleShortInfo], c: CompletedRasterGrid):
# if s is None:
# return
#
# try:
# response = self.__proc_api.create_result(
#
# )
print(e)