From e4db694c39d7cbaf81cdb2c1684aeb6b1eb9065d Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Wed, 1 Oct 2025 17:11:06 +0200 Subject: [PATCH] DAQ: and models.py - added last_best_res calculation and value to daq and models --- common/src/aaredaqlib/models.py | 1 + daq/src/aaredaq/daq.py | 124 ++++++++++++++++++++------------ 2 files changed, 81 insertions(+), 44 deletions(-) diff --git a/common/src/aaredaqlib/models.py b/common/src/aaredaqlib/models.py index db7b1972..c14bff9b 100644 --- a/common/src/aaredaqlib/models.py +++ b/common/src/aaredaqlib/models.py @@ -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 diff --git a/daq/src/aaredaq/daq.py b/daq/src/aaredaq/daq.py index cba37a60..bca90752 100644 --- a/daq/src/aaredaq/daq.py +++ b/daq/src/aaredaq/daq.py @@ -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: