DAQ/Server: added auto_exposure end point, removed commented out code from autofocus and started a blower_control function
This commit is contained in:
+8
-120
@@ -601,6 +601,12 @@ class AareDAQ:
|
||||
logger.error(f"Failed to park and dry: {e}")
|
||||
raise
|
||||
|
||||
def blower_control(self):
|
||||
try:
|
||||
self.__devs.tell.toggle_blower()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to turn off blower: {e}")
|
||||
|
||||
def __magnet_position_sensor_check(self, timeout: float = 1.0, repeat: bool = True):
|
||||
#TODO check this works, add beamstop z controls and test.
|
||||
|
||||
@@ -710,126 +716,6 @@ class AareDAQ:
|
||||
Best Z position (mm) found during the scan
|
||||
"""
|
||||
raise NotImplementedError("Autofocus not implemented yet")
|
||||
geom = self.sample_geometry
|
||||
current_smargon = self.__devs.smargon_pos
|
||||
|
||||
# Get center for the mask (use beam center if not specified)
|
||||
center_x = geom.beam_location_pxl.x
|
||||
center_y = geom.beam_location_pxl.y
|
||||
radius_pxl = 30
|
||||
|
||||
# Convert z_range from um to mm
|
||||
z_range_mm = settings.z_range_um / 1000.0
|
||||
n_steps = settings.z_steps
|
||||
|
||||
# Starting Z position (current sh_mm)
|
||||
z_start = 0.0
|
||||
z_min = z_start - z_range_mm / 2.0
|
||||
z_max = z_start + z_range_mm / 2.0
|
||||
z_step = z_range_mm / (n_steps - 1) if n_steps > 1 else 0.0
|
||||
|
||||
# Pre-compute mask (will be created on first image)
|
||||
focus_mask: np.ndarray | None = None
|
||||
|
||||
# Collect focus measures at each Z position
|
||||
z_positions: list[float] = []
|
||||
focus_values: list[float] = []
|
||||
|
||||
print(f"Starting autofocus: z_range={z_range_mm * 1000:.1f}um, steps={n_steps}, "
|
||||
f"center=({center_x:.1f}, {center_y:.1f}), radius={radius_pxl:.1f}px")
|
||||
|
||||
|
||||
for i in range(n_steps):
|
||||
z_pos = z_min + i * z_step
|
||||
|
||||
# Move to position
|
||||
target = SmargonCoordinate(
|
||||
chi_deg=current_smargon.chi_deg,
|
||||
phi_deg=current_smargon.phi_deg,
|
||||
sh_mm=geom.beamline_to_smargon(Coordinate(z=z_pos))
|
||||
)
|
||||
self.__devs.smargon_pos = target
|
||||
self.__devs.smargon_wait(timeout=30)
|
||||
|
||||
# Wait for mechanical settling and image stabilization
|
||||
time.sleep(settle_time_s)
|
||||
|
||||
# Capture image
|
||||
gray = self.camera_image_gray
|
||||
|
||||
if gray is None:
|
||||
logger.warning(f"Failed to get image at z={z_pos:.4f}")
|
||||
continue
|
||||
|
||||
# Create mask on first valid image
|
||||
if focus_mask is None or focus_mask.shape != gray.shape:
|
||||
h, w = gray.shape
|
||||
y, x = np.ogrid[:h, :w]
|
||||
focus_mask = (x - center_x) ** 2 + (y - center_y) ** 2 <= radius_pxl ** 2
|
||||
|
||||
print(focus_mask.shape)
|
||||
print(gray.shape)
|
||||
|
||||
# Calculate focus measure
|
||||
fm = focus_measure_edges(gray, focus_mask)
|
||||
|
||||
z_positions.append(z_pos)
|
||||
focus_values.append(fm)
|
||||
logger.debug(f"Autofocus step {i + 1}/{n_steps}: z={z_pos:.4f}mm, focus={fm:.2f}")
|
||||
print(f"Autofocus step {i + 1}/{n_steps}: z={z_pos:.4f}mm, focus={fm:.2f}")
|
||||
if len(z_positions) < 3:
|
||||
logger.error("Autofocus failed: not enough valid measurements")
|
||||
# Return to original position
|
||||
self.__devs.smargon_pos = current_smargon
|
||||
self.__devs.smargon_wait(timeout=30)
|
||||
return z_start
|
||||
|
||||
# Find best position - use parabolic fit around the peak for sub-step precision
|
||||
z_arr = np.array(z_positions)
|
||||
fm_arr = np.array(focus_values)
|
||||
|
||||
# Find index of maximum
|
||||
peak_idx = int(np.argmax(fm_arr))
|
||||
|
||||
# Try parabolic fit if peak is not at the edge
|
||||
if 0 < peak_idx < len(fm_arr) - 1:
|
||||
# Fit parabola to 3 points around peak: f(z) = a*z^2 + b*z + c
|
||||
z_fit = z_arr[peak_idx - 1: peak_idx + 2]
|
||||
fm_fit = fm_arr[peak_idx - 1: peak_idx + 2]
|
||||
try:
|
||||
coeffs = np.polyfit(z_fit, fm_fit, 2)
|
||||
a, b, c = coeffs
|
||||
if a < 0: # Parabola opens downward (valid peak)
|
||||
best_z = -b / (2 * a)
|
||||
# Sanity check: best_z should be within the fitted range
|
||||
if z_fit[0] <= best_z <= z_fit[2]:
|
||||
logger.info(f"Autofocus: parabolic fit found peak at z={best_z:.4f}mm")
|
||||
else:
|
||||
best_z = z_arr[peak_idx]
|
||||
logger.info(f"Autofocus: parabolic fit out of range, using sample peak z={best_z:.4f}mm")
|
||||
else:
|
||||
best_z = z_arr[peak_idx]
|
||||
logger.info(f"Autofocus: invalid parabola, using sample peak z={best_z:.4f}mm")
|
||||
except Exception as e:
|
||||
logger.warning(f"Parabolic fit failed: {e}, using sample peak")
|
||||
best_z = z_arr[peak_idx]
|
||||
else:
|
||||
best_z = z_arr[peak_idx]
|
||||
logger.warning(f"Autofocus: peak at edge of scan range, z={best_z:.4f}mm")
|
||||
|
||||
# Move to best position
|
||||
best_target = SmargonCoordinate(
|
||||
chi_deg=current_smargon.chi_deg,
|
||||
phi_deg=current_smargon.phi_deg,
|
||||
sh_mm=geom.beamline_to_smargon(Coordinate(z=best_z)),
|
||||
)
|
||||
self.__devs.smargon_pos = best_target
|
||||
self.__devs.smargon_wait(timeout=30)
|
||||
|
||||
logger.warning(f"Autofocus complete: best_z={best_z:.4f}mm, "
|
||||
f"focus_range=[{min(fm_arr):.2f}, {max(fm_arr):.2f}]")
|
||||
|
||||
return best_z
|
||||
|
||||
def auto_focus(self, settings: AutofocusSettings) -> float:
|
||||
"""
|
||||
@@ -851,6 +737,8 @@ class AareDAQ:
|
||||
self.__cfg.state_busy = False
|
||||
raise
|
||||
|
||||
def auto_exposure(self):
|
||||
self.__devs.samcam_auto(AutoEnum.ONCE)
|
||||
|
||||
def __auto_center(self, request: RasterGridRequest) -> CompletedRasterGrid | None:
|
||||
#TODO do we need to handle the two grid scans differently?
|
||||
|
||||
@@ -543,6 +543,23 @@ async def samcam_settings(s: SampleCameraSettings, token: str = Depends(oauth2_s
|
||||
daq.samcam_settings = s
|
||||
return "OK"
|
||||
|
||||
@app.put("/beamline/autoexposure")
|
||||
async def samcam_autoexposure(token: str = Depends(oauth2_scheme)):
|
||||
"""
|
||||
Update the sample camera settings (exposure, gain, etc.).
|
||||
|
||||
Args:
|
||||
s: SampleCameraSettings object.
|
||||
token: OAuth2 access token.
|
||||
|
||||
Returns:
|
||||
"OK" on success.
|
||||
"""
|
||||
logger.debug(f"SamCam settings: {s}")
|
||||
auth.check_jwt_rw(cfg, auth.parse_token(token))
|
||||
daq.auto_exposure()
|
||||
return "OK"
|
||||
|
||||
@app.post("/samcam/autofocus")
|
||||
async def samcam_autofocus(s: AutofocusSettings, token: str = Depends(oauth2_scheme)):
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user