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

DAQ: and models.py - added last_best_res calculation and value to daq and models

Browse Source
This commit is contained in:
leonarski_f
2025-10-01 17:11:06 +02:00
parent 19d5740c1f
commit e4db694c39
2 changed files with 81 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/src/aaredaqlib/models.py
+1
View File
@@ -601,6 +601,7 @@ class DAQStatusModel(BaseModel):
sample: SampleShortInfo | None = None
session: SessionStatus
box: BoundingBoxModel | None = None
last_best_res: float | None = None
class BeamlineSettingsModel(BaseModel):
dtz_max: float | None = 1600.0
daq/src/aaredaq/daq.py
+80 -44
View File
@@ -577,6 +577,75 @@ class AareDAQ:
else:
return None
def raster_centre_of_mass(self, images, r:RasterGridRequest, result):
# if any(img.index for img in images):
# print("COM by indexed spots")
# result_array= self.create_quality_filtered_array(images, 'spots_indexed', min_spots=None,
# min_efficiency=1.0, min_background=None)
# else:
print("COM by low res spots")
result_array = self.create_quality_filtered_array(images, 'spots_low_res', min_spots=None,
min_efficiency=1.0, min_background=None)
if r.n_x == 1:
print('vertical scan')
max_image = max(images, key=lambda img: img.spots_low_res)
new_y = max_image.ny
com = (0, new_y)
else:
print('horizontal scan')
com = ndimage.center_of_mass(result_array)
print(f"Center of mass: {com}")
if np.isnan(com[1]) or np.isnan(com[0]):
print("Center of mass is nan")
com = None
grid_mm_x = None
grid_mm_y = None
best_res = None
else:
grid_mm_x = com[0] * r.grid_size_mm.x
grid_mm_y = com[1] * r.grid_size_mm.y
if r.n_x == 1:
grid_mm_x += (0.5 * r.grid_size_mm.x)
cx, cy = com[::-1] # com=(y, x) -> (x, y), rounded once
start_x, end_x = round(cx - 1), round(cx + 1)
start_y, end_y = round(cy - 1), round(cy + 1)
# Collect images in the 3x3 neighborhood around the center
res_list = [img for img in images
if start_x <= img.nx <= end_x and start_y <= img.ny <= end_y]
# Best by res, skipping None
best_res = max((img for img in res_list if img.res is not None),
key=lambda img: img.res,
default=None)
for image in images:
if image.nx == round(com[0]) and image.ny == round(com[1]):
print(f"com found for image: {image.number}")
try:
if best_res is not None and best_res.res is not None:
# store the numeric resolution on the config/session so it appears in status
self.__cfg.last_best_res = float(best_res.res)
else:
self.__cfg.last_best_res = None
except Exception as e:
# Don't break the flow if best_res cannot be published
self.__cfg.last_best_res = None
if com is not None and grid_mm_x is not None and grid_mm_y is not None:
new_delta_mm = self.sample_geometry.smargon_nudge(Coordinate(x=grid_mm_x, y=grid_mm_y))
print(f"new delta mm: {new_delta_mm}, new grid x: {grid_mm_x}, new grid y: {grid_mm_y}")
else:
print(f"using old method as COM is none or nan")
new_delta_mm = self.identify_crystal_raster(result, r)
return new_delta_mm, best_res
def __raster(self, r: RasterGridRequest) -> CompletedRasterGridElem:
max_time = r.exp_time_s * r.n_y * r.n_x + 60
if r.dtz is not None:
@@ -623,53 +692,19 @@ class AareDAQ:
images = result.images
output_data = {
'timestamp': time.ctime(),
'scan_results': [result.model_dump() for result in images],
'total_results': len([result for result in images])
'scan_results': [image.model_dump() for image in images],
'total_results': len([image for image in images])
}
if self.sample is not None and self.sample.db_id is not None:
with open(f'{self.sample.db_id}_scan_results.json', 'w') as f:
json.dump(output_data, f, indent=2)
# if any(img.index for img in images):
# print("COM by indexed spots")
# result_array= self.create_quality_filtered_array(images, 'spots_indexed', min_spots=None,
# min_efficiency=1.0, min_background=None)
# else:
print("COM by low res spots")
result_array = self.create_quality_filtered_array(images, 'spots_low_res', min_spots=None,
min_efficiency=1.0, min_background=None)
if r.n_x == 1:
print('vertical scan')
max_image = max(images, key=lambda img: img.spots_low_res)
new_y = max_image.ny
com = (0, new_y)
else:
print('horizontal scan')
com = ndimage.center_of_mass(result_array)
print(f"Center of mass: {com}")
if np.isnan(com[1]) or np.isnan(com[0]):
print("Center of mass is nan")
com = None
grid_mm_x=None
grid_mm_y=None
else:
grid_mm_x = com[0] * r.grid_size_mm.x
grid_mm_y = com[1] * r.grid_size_mm.y
if r.n_x == 1:
grid_mm_x += (0.5 * r.grid_size_mm.x)
for image in images:
if image.nx == round(com[0]) and image.ny == round(com[1]):
print(f"com found for image: {image.number}")
if com is not None and grid_mm_x is not None and grid_mm_y is not None:
new_delta_mm = self.sample_geometry.smargon_nudge(Coordinate(x=grid_mm_x, y=grid_mm_y))
print(f"new delta mm: {new_delta_mm}, new grid x: {grid_mm_x}, new grid y: {grid_mm_y}")
else:
print(f"using old method as COM is none or nan")
new_delta_mm = self.identify_crystal_raster(result, r)
filename=f'{self.sample.db_id}_scan_results_vertical.json'
else:
filename=f'{self.sample.db_id}_scan_results_horizontal.json'
with open(filename, 'w') as f:
json.dump(output_data, f, indent=2)
print('before centre_of_mass')
new_delta_mm, best_res = self.raster_centre_of_mass(images, r, result)
print('after centre_of_mass')
if new_delta_mm is not None:
print(f'{time.ctime()}, moving SMARGON to target new delta mm {r.smargon.sh_mm + new_delta_mm} mm')
self.__devs.smargon.target = SmargonCoordinate(sh_mm=r.smargon.sh_mm + new_delta_mm,
@@ -678,6 +713,7 @@ class AareDAQ:
else:
print("Auto finding optimal image failed due to no images found. Using previous position.")
self.__devs.smargon.target = save_smargon_position
return CompletedRasterGridElem(request=copy.deepcopy(r), result=result)
def measure_raster(self, r: RasterGridRequest, auto: bool) -> CompletedRasterGrid: