Added par_spot_selection_mode

Author: Andrea-gm

autoemx/config/runtime_configs.py

@@ -26,7 +26,7 @@
 """
 import numpy as np
 import multiprocessing
-from typing import Any, ClassVar, Dict, List, Optional, Tuple
+from typing import Any, ClassVar, Dict, List, Literal, Optional, Tuple
 
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
@@ -345,7 +345,7 @@ class PowderMeasurementConfig(BaseModel):
     Configuration for powder measurement.
 
     Attributes:
-        is_manual_particle_selection (bool): Whether to manually navigate sample to select particles to analyse (Default = False).
+        par_selection_mode (str): 'auto' for automatic particle navigation, 'manual' to prompt user to center each particle (default: 'auto').
         is_known_powder_mixture_meas (bool): Whether sample is a known binary mixture of powders. Used to characterize precursor extent of intermixing (Default = False).
         img_shift_tracking (bool): Whether to use image shift tracking during acquisition (Default = True).
         par_search_frame_width_um (float, optional): Frame width used when searching for particles, in um.
@@ -365,10 +365,13 @@ class PowderMeasurementConfig(BaseModel):
             Particle pixel intensities are scaled to 8-bit prior threhsolding, i.e., darkest pixel will be set to 0, and brightest to 255.
         par_feature_selection (str): 'random' for random selection of points within bright regions, 'peaks' for brightest peak spots (default: 'random').
         par_spot_spacing (str): 'random' for unbiased spot selecton, 'maximized' for maximized spot spacing over particle (default: 'random').
+        par_spot_selection_mode (str): 'auto' for built-in spot selection, 'callback' to supply spots via xsp_spot_selector (default: 'auto').
     """
     DEFAULT_PAR_SEGMENTATION_MODEL: ClassVar[str] = "threshold_bright"
+    AVAILABLE_SPOT_SELECTION_MODES: ClassVar[Tuple[str, ...]] = ('auto', 'callback')
+    AVAILABLE_PAR_SELECTION: ClassVar[Tuple[str, ...]] = ('auto', 'manual')
 
-    is_manual_particle_selection: bool = False
+    par_selection_mode: Literal['auto', 'manual'] = 'auto'
     is_known_powder_mixture_meas: bool = False
     img_shift_tracking: bool = True
     par_search_frame_width_um: Optional[float] = None
@@ -378,7 +381,8 @@ class PowderMeasurementConfig(BaseModel):
     min_area_par: float = 10.0     # µm²
     par_mask_margin: float = 1.0   # µm
     xsp_spots_distance_um: float = 1.0  # µm
-    par_segmentation_model: str = "threshold_bright"
+    par_spot_selection_mode: Literal['auto', 'callback'] = 'auto'
+    par_segmentation_model: str = DEFAULT_PAR_SEGMENTATION_MODEL    # "threshold_bright"    
     par_brightness_thresh: int = 100  # in 8-bit image
     par_xy_spots_thresh: int = 100
     par_feature_selection: str = 'random'
@@ -390,6 +394,17 @@ class PowderMeasurementConfig(BaseModel):
 
     model_config = ConfigDict(extra="forbid")
 
+    @model_validator(mode="before")
+    @classmethod
+    def _normalize_legacy_fields(cls, data: Any) -> Any:
+        if isinstance(data, dict):
+            cleaned = dict(data)
+            legacy_manual = cleaned.pop("is_manual_particle_selection", None)
+            if legacy_manual is not None and "par_selection_mode" not in cleaned:
+                cleaned["par_selection_mode"] = "manual" if legacy_manual else "auto"
+            return cleaned
+        return data
+
     @model_validator(mode="after")
     def _validate(self) -> "PowderMeasurementConfig":
         if self.par_segmentation_model not in self.AVAILABLE_PAR_SEGMENTATION_MODELS:
@@ -407,6 +422,16 @@ def _validate(self) -> "PowderMeasurementConfig":
                 f'Value of "par_spot_spacing" set to {self.par_spot_spacing} is invalid. '
                 f'Must be one of {self.AVAILABLE_SPOT_SPACING_SELECTION}.'
             )
+        if self.par_spot_selection_mode not in self.AVAILABLE_SPOT_SELECTION_MODES:
+            raise ValueError(
+                f'Value of "par_spot_selection_mode" set to {self.par_spot_selection_mode} is invalid. '
+                f'Must be one of {self.AVAILABLE_SPOT_SELECTION_MODES}.'
+            )
+        if self.par_selection_mode not in self.AVAILABLE_PAR_SELECTION:
+            raise ValueError(
+                f'Value of "par_selection_mode" set to {self.par_selection_mode} is invalid. '
+                f'Must be one of {self.AVAILABLE_PAR_SELECTION}.'
+            )
         if self.min_area_par < 0 or self.max_area_par < 0:
             raise ValueError("Particle area thresholds must be non-negative.")
         if self.max_area_par < self.min_area_par:

autoemx/core/composition_analysis/analyser.py

@@ -79,6 +79,7 @@
 # Project-specific imports
 from autoemx.core.quantifier import XSp_Quantifier
 from autoemx.core.em_runtime.controller import EM_Controller
+from autoemx.core.em_runtime.xsp_spot_selection import XSpSpotSelectorCallback
 from autoemx.core.em_runtime.sample_finder import EM_Sample_Finder
 import autoemx.calibrations as calibs
 import autoemx.utils.constants as cnst
@@ -522,6 +523,7 @@ def __init__(
         verbose: bool = True,
         results_dir: Optional[str] = None,
         sample_id: Optional[str] = None,
+        xsp_spot_selector: Optional[XSpSpotSelectorCallback] = None,
     ):
         """
         Initialize the EMXSp_Composition_Analyzer with all configuration objects.
@@ -544,6 +546,7 @@ def __init__(
         # --- Define use of class instance
         self.is_acquisition = is_acquisition
         self.development_mode = development_mode
+        self.xsp_spot_selector = xsp_spot_selector
 
         measurement_cfg = measurement_cfg.model_copy(
             update={
@@ -806,6 +809,7 @@ def _initialise_SEM(self) -> None:
             sample_id=self.sample_id,
             results_dir=EM_images_dir,
             verbose=self.verbose,
+            xsp_spot_selector=self.xsp_spot_selector,
         )
         self.EM_controller.initialise_SEM()
         self.EM_controller.initialise_sample_navigator(exclude_sample_margin=True)
@@ -1127,7 +1131,7 @@ def _save_analysis_config_summary(self) -> None:
             with open(summary_path, "w", encoding="utf-8") as fh:
                 fh.write("\n".join(lines))
         except OSError as e:
-            logging.warning(f"Could not write analysis config summary: {e}")
+            logger.warning(f"Could not write analysis config summary: {e}")
 
 
     def _resolve_active_analysis_config_ids(self) -> Tuple[int, int]:

autoemx/core/em_runtime/controller.py

@@ -59,6 +59,7 @@
 from autoemx.core.em_runtime.frame_navigator import FrameNavigator
 from autoemx.core.em_runtime.spectrum_acquisition import SpectrumAcquisition
 from autoemx.core.em_runtime import image_utilities
+from autoemx.core.em_runtime.xsp_spot_selection import XSpSpotSelectorCallback
 
 from autoemx._logging import get_logger
 logger = get_logger(__name__)
@@ -135,6 +136,7 @@ def __init__(
         results_dir: Optional[str] = None,
         verbose: bool = True,
         development_mode: bool = False,
+        xsp_spot_selector: Optional[XSpSpotSelectorCallback] = None,
     ):
         """
         Initialize an EM_Controller object from a ``LedgerConfigs`` bundle.
@@ -245,7 +247,8 @@ def __init__(
             EM_driver_obj=EM_driver,
             results_dir=results_dir,
             verbose=verbose,
-            development_mode=development_mode
+            development_mode=development_mode,
+            xsp_spot_selector=xsp_spot_selector,
         )
 
         self.spectrum_acq = SpectrumAcquisition(
@@ -274,6 +277,7 @@ def from_configs(
         results_dir: Optional[str] = None,
         verbose: bool = True,
         development_mode: Optional[bool] = False,
+        xsp_spot_selector: Optional[XSpSpotSelectorCallback] = None,
     ) -> "EM_Controller":
         """
         Construct an EM_Controller from individual config objects.
@@ -317,6 +321,7 @@ def from_configs(
             results_dir=results_dir,
             verbose=verbose,
             development_mode=development_mode,
+            xsp_spot_selector=xsp_spot_selector,
         )
 
     #%% Microscope initialization

autoemx/core/em_runtime/frame_navigator.py

@@ -14,9 +14,13 @@
 Created on 2026
 @author: Andrea
 """
+from typing import Optional
+
 import time
 import numpy as np
 
+from autoemx.core.em_runtime.xsp_spot_selection import XSpSpotSelectorCallback
+
 import autoemx.utils.constants as cnst
 from autoemx.utils.helper import AlphabetMapper, Prompt_User, print_single_separator
 from autoemx.core.em_runtime.particle_finder import EM_Particle_Finder
@@ -68,7 +72,8 @@ class FrameNavigator:
     def __init__(self, EM_controller, sample_cfg, sample_substrate_cfg, measurement_cfg,
                  powder_meas_cfg, bulk_meas_cfg, center_pos, sample_hw_mm,
                  im_width, im_height, EM_driver_obj, results_dir=None,
-                 verbose=True, development_mode=False):
+                 verbose=True, development_mode=False,
+                 xsp_spot_selector: Optional[XSpSpotSelectorCallback] = None):
         self.EM_controller = EM_controller
         self.sample_cfg = sample_cfg
         self.sample_substrate_cfg = sample_substrate_cfg
@@ -83,6 +88,7 @@ def __init__(self, EM_controller, sample_cfg, sample_substrate_cfg, measurement_
         self.results_dir = results_dir
         self.verbose = verbose
         self.development_mode = development_mode
+        self.xsp_spot_selector = xsp_spot_selector
 
         # Frame tracking
         self._frame_cntr = 0
@@ -143,7 +149,7 @@ def initialise_sample_navigator(self, grid_search_fw_mm, exclude_sample_margin=T
             self.particle_finder = EM_Particle_Finder(
                 self.EM_controller,
                 powder_meas_cfg=self.powder_meas_cfg,
-                is_manual_particle_selection=self.powder_meas_cfg.is_manual_particle_selection,
+                xsp_spot_selector=self.xsp_spot_selector,
                 results_dir=self.results_dir,
                 verbose=self.verbose,
                 development_mode=self.development_mode

autoemx/core/em_runtime/particle_finder.py

@@ -72,6 +72,12 @@
 
 from typing import List, Optional
 
+from autoemx.core.em_runtime.xsp_spot_selection import (
+    XSpSpotSelectionContext,
+    XSpSpotSelectorCallback,
+    validate_xsp_spot_pixels,
+)
+
 # Local project imports
 from autoemx.utils.helper import (
     Prompt_User,
@@ -134,8 +140,8 @@ class EM_Particle_Finder:
         Reference to the parent EM_Controller instance (must be initialised before use).
     powder_meas_cfg : PowderMeasurementConfig
         Configuration object for powder measurement (see dataclass for details).
-    is_manual_particle_selection : bool
-        If True, prompts user to center image around next particle to analyse.
+    par_selection_mode : str
+        Particle navigation mode from ``powder_meas_cfg`` ('auto' or 'manual').
     results_dir : str or None
         Directory for saving result images and data.
     verbose : bool
@@ -167,7 +173,7 @@ def __init__(
         self,
         EM_controller,
         powder_meas_cfg: PowderMeasurementConfig,
-        is_manual_particle_selection: bool = False,
+        xsp_spot_selector: Optional[XSpSpotSelectorCallback] = None,
         results_dir: Optional[str] = None,
         verbose: bool = True,
         development_mode: bool = True
@@ -181,10 +187,10 @@ def __init__(
         ----------
         EM_controller : EM_Controller
             Reference to the parent EM_Controller instance (must be initialised before use).
-        powder_meas_cfg : PowderMeasurementConfig, optional
+        powder_meas_cfg : PowderMeasurementConfig
             Configuration object for powder measurement (see dataclass for details).
-        is_manual_particle_selection : bool, optional
-            If True, enables manual particle selection mode (default: False).
+        xsp_spot_selector : callable, optional
+            Required when ``powder_meas_cfg.par_spot_selection_mode='callback'``.
         results_dir : str, optional
             Directory to save result images and data (default: None).
         verbose : bool, optional
@@ -213,7 +219,11 @@ def __init__(
             self._im_width = EM_controller.im_width
             self._im_height = EM_controller.im_height
 
-        self.is_manual_particle_selection = is_manual_particle_selection
+        self.xsp_spot_selector = xsp_spot_selector
+        if self.powder_meas_cfg.par_spot_selection_mode == 'callback' and self.xsp_spot_selector is None:
+            raise ValueError(
+                "powder_meas_cfg.par_spot_selection_mode='callback' requires xsp_spot_selector."
+            )
         # NOTE: self.tot_par_cntr is initialized below, so _select_par_prompt_title will be set on first use
         if self.is_manual_particle_selection:
             self._select_par_prompt_title = f"Select position for particle #{self.tot_par_cntr}"
@@ -234,7 +244,16 @@ def __init__(
         self._num_par_in_frame = 0
         self.analyzed_pars: List[tuple(float, str)] = []
 
+    @property
+    def is_manual_particle_selection(self) -> bool:
+        """Whether particle navigation uses manual centering prompts."""
+        return self.powder_meas_cfg.par_selection_mode == 'manual'
 
+    @property
+    def par_selection_mode(self) -> str:
+        return self.powder_meas_cfg.par_selection_mode
+
+    
     def _check_EM_controller_initialization(self) -> None:
         """
         Check whether the associated EM_Controller instance is initialized.
@@ -987,6 +1006,14 @@ def get_XS_acquisition_spots_coord_list(
         # --- 2. Erode the particle mask to avoid edge effects ---
         margin = max(10, int(self.powder_meas_cfg.par_mask_margin / self.EM.pixel_size_um))
         final_mask = self._erode_particle_mask(par_mask, margin)
+
+        if self.powder_meas_cfg.par_spot_selection_mode == 'callback':
+            return self._get_callback_xsp_spots(
+                n_tot_sp_collected=n_tot_sp_collected,
+                par_image=par_image,
+                par_mask=par_mask,
+                usable_mask=final_mask,
+            )
 
         # --- 3. Find bright points in image, which indicate highest regions on particle---
         thresholded_image, min_area_pixels = self._find_particle_bright_regions(final_mask, par_image)
@@ -1027,6 +1054,48 @@ def get_XS_acquisition_spots_coord_list(
         self.ref_image = par_image
         return [tuple(map(int, pt)) for pt in selected_points]
 
+    def _get_callback_xsp_spots(
+        self,
+        n_tot_sp_collected: int,
+        par_image,
+        par_mask,
+        usable_mask,
+    ) -> List[tuple]:
+        """Invoke xsp_spot_selector and validate returned pixel coordinates."""
+        frame_label = getattr(self.EM, 'current_frame_label', '')
+        if frame_label is None:
+            frame_label = ''
+
+        context = XSpSpotSelectionContext(
+            particle_id=self.tot_par_cntr,
+            n_tot_sp_collected=n_tot_sp_collected,
+            par_image=par_image,
+            par_mask=par_mask,
+            usable_mask=usable_mask,
+            pixel_size_um=float(self.EM.pixel_size_um),
+            frame_label=str(frame_label),
+            im_width=int(self._im_width),
+            im_height=int(self._im_height),
+            particle_finder=self,
+        )
+        try:
+            raw_spots = self.xsp_spot_selector(context)
+        except Exception as exc:
+            logger.error(f"❌ xsp_spot_selector callback failed: {exc}")
+            raise
+
+        if not raw_spots:
+            self.ref_image = par_image
+            return []
+
+        selected_points = validate_xsp_spot_pixels(
+            raw_spots,
+            im_width=int(self._im_width),
+            im_height=int(self._im_height),
+        )
+        self.ref_image = par_image
+        return selected_points
+
 
     def _collect_candidate_points(self, thresholded_image, par_image, feature_selection, min_area_pixels):
         """