import os
import configparser
import cv2
import h5py
import matplotlib.pyplot as plt
import numpy as np
from matplotlib import colors
from PIL import Image, ImageDraw, ImageEnhance
from PIL.ImageQt import ImageQt
from PyQt6.QtCore import QCoreApplication, QPointF, QRectF, Qt, QTimer
from PyQt6.QtGui import QColor, QFont, QPainter, QPainterPath, QPen, QPixmap
from PyQt6.QtWidgets import QMessageBox
from coseda.logging_utils import log_print
from coseda.io import config_to_paths
from coseda.nexus.paths import get_mask_dataset
from coseda.peakfinding.findpeaks import process_single_frame
from coseda.peakfinding.maxres import write_maxres_dataset, write_maxres_from_config
from coseda.peakfinding.peakfinder9 import find_peaks_pf9_multiscale
from cosedaUI import AuroraPlotWindow, MapWindow
from cosedaUI.image_tools import (
export_display_image, export_display_pixmap, export_raw_image, draw_pixmap_scalebar,
)
[docs]
class ViewerMixin:
def _clear_strong_peaks_order(self):
self._strong_peaks_order = []
self._strong_peaks_rank = {}
def _clear_maxres_order(self):
self._maxres_order = []
self._maxres_rank = {}
def _ensure_strong_peaks_order(self):
"""Build frame order sorted by nPeaks descending (tie-break by frame index)."""
if self.total_frames <= 0:
self._clear_strong_peaks_order()
return False
if self._strong_peaks_order and len(self._strong_peaks_order) == self.total_frames:
return True
if not self.hdf5_path or not os.path.exists(self.hdf5_path):
self._clear_strong_peaks_order()
return False
try:
with h5py.File(self.hdf5_path, 'r') as file:
if 'entry/data/nPeaks' not in file:
self._clear_strong_peaks_order()
return False
n_peaks_ds = file['entry/data/nPeaks']
npeaks_total = min(self.total_frames, int(n_peaks_ds.shape[0]))
if npeaks_total <= 0:
self._clear_strong_peaks_order()
return False
n_peaks = np.asarray(n_peaks_ds[:npeaks_total], dtype=np.float64).reshape(-1)
order = np.lexsort((np.arange(npeaks_total, dtype=np.int64), -n_peaks))
ordered_frames = [int(i) for i in order.tolist()]
# Keep all frames addressable even if nPeaks length is shorter than total_frames.
if npeaks_total < self.total_frames:
ordered_frames.extend(range(npeaks_total, self.total_frames))
self._strong_peaks_order = ordered_frames
self._strong_peaks_rank = {
int(frame_idx): int(pos)
for pos, frame_idx in enumerate(self._strong_peaks_order)
}
return len(self._strong_peaks_order) > 0
except Exception as exc:
log_print(f"Error preparing strong-peaks ordering: {exc}")
self._clear_strong_peaks_order()
return False
def _ensure_maxres_dataset(self):
if not self.hdf5_path or not os.path.exists(self.hdf5_path):
return False
try:
with h5py.File(self.hdf5_path, 'r') as file:
if 'entry/data/maxres' in file:
return True
except Exception:
return False
try:
with self._with_busy_status("Computing max resolution per frame..."):
if self.full_ini_file_path and os.path.exists(self.full_ini_file_path):
write_maxres_from_config(
configfile=self.full_ini_file_path,
logfile_path=self.log_file_path,
)
else:
write_maxres_dataset(
h5file_path=self.hdf5_path,
center_x="h5",
center_y="h5",
logfile_path=self.log_file_path,
)
except Exception as exc:
log_print(f"Error preparing maxres dataset: {exc}")
QMessageBox.warning(
self,
"Highest Resolution Unavailable",
f"Could not generate maxres dataset:\n{exc}",
)
return False
try:
with h5py.File(self.hdf5_path, 'r') as file:
return 'entry/data/maxres' in file
except Exception:
return False
def _ensure_maxres_order(self, auto_generate=False):
"""Build frame order sorted by maxres descending (tie-break by frame index)."""
if self.total_frames <= 0:
self._clear_maxres_order()
return False
if self._maxres_order and len(self._maxres_order) == self.total_frames:
return True
if not self.hdf5_path or not os.path.exists(self.hdf5_path):
self._clear_maxres_order()
return False
if auto_generate and not self._ensure_maxres_dataset():
self._clear_maxres_order()
return False
try:
with h5py.File(self.hdf5_path, 'r') as file:
if 'entry/data/maxres' not in file:
self._clear_maxres_order()
return False
maxres_ds = file['entry/data/maxres']
maxres_total = min(self.total_frames, int(maxres_ds.shape[0]))
if maxres_total <= 0:
self._clear_maxres_order()
return False
maxres_vals = np.asarray(maxres_ds[:maxres_total], dtype=np.float64).reshape(-1)
order = np.lexsort((np.arange(maxres_total, dtype=np.int64), -maxres_vals))
ordered_frames = [int(i) for i in order.tolist()]
if maxres_total < self.total_frames:
ordered_frames.extend(range(maxres_total, self.total_frames))
self._maxres_order = ordered_frames
self._maxres_rank = {
int(frame_idx): int(pos)
for pos, frame_idx in enumerate(self._maxres_order)
}
return len(self._maxres_order) > 0
except Exception as exc:
log_print(f"Error preparing maxres ordering: {exc}")
self._clear_maxres_order()
return False
def _is_show_strong_peaks_active(self):
return (
hasattr(self, 'show_strong_peaks_checkbox')
and self.show_strong_peaks_checkbox.isEnabled()
and self.show_strong_peaks_checkbox.isChecked()
)
def _is_show_maxres_active(self):
return bool(getattr(self, '_maxres_order_active', False))
def _set_strong_peaks_option_available(self, available):
if not hasattr(self, 'frame_order_combo'):
return
model = self.frame_order_combo.model()
item = model.item(1) if hasattr(model, 'item') else None
if item is not None:
item.setEnabled(bool(available))
if not available and self.frame_order_combo.currentIndex() == 1:
self.frame_order_combo.blockSignals(True)
self.frame_order_combo.setCurrentIndex(0)
self.frame_order_combo.blockSignals(False)
def _set_maxres_option_available(self, available):
if not hasattr(self, 'frame_order_combo'):
return
model = self.frame_order_combo.model()
item = model.item(2) if hasattr(model, 'item') else None
if item is not None:
item.setEnabled(bool(available))
if not available:
self._maxres_order_active = False
self._clear_maxres_order()
if not available and self.frame_order_combo.currentIndex() == 2:
self.frame_order_combo.blockSignals(True)
self.frame_order_combo.setCurrentIndex(0)
self.frame_order_combo.blockSignals(False)
def _sync_frame_order_combo_with_mode(self):
if not hasattr(self, 'frame_order_combo'):
return
if self._is_show_maxres_active():
target_index = 2
elif self._is_show_strong_peaks_active():
target_index = 1
else:
target_index = 0
if self.frame_order_combo.currentIndex() == target_index:
return
self.frame_order_combo.blockSignals(True)
self.frame_order_combo.setCurrentIndex(target_index)
self.frame_order_combo.blockSignals(False)
def _activate_maxres_order_mode(self):
if not self._ensure_maxres_order(auto_generate=True):
self._maxres_order_active = False
self._sync_frame_order_combo_with_mode()
return
self._maxres_order_active = True
if hasattr(self, 'show_strong_peaks_checkbox'):
self.show_strong_peaks_checkbox.blockSignals(True)
self.show_strong_peaks_checkbox.setChecked(False)
self.show_strong_peaks_checkbox.blockSignals(False)
if self._maxres_order:
self.current_frame_index = int(self._maxres_order[0])
self._set_frame_slider_range_for_mode()
self._sync_frame_slider_with_current_frame()
self._sync_frame_order_combo_with_mode()
self.update_image()
[docs]
def on_frame_order_changed(self, index):
if not hasattr(self, 'show_strong_peaks_checkbox'):
return
if index == 0:
changed_mode = self._is_show_maxres_active()
self._maxres_order_active = False
if self.show_strong_peaks_checkbox.isChecked():
self.show_strong_peaks_checkbox.setChecked(False)
return
if changed_mode:
self._set_frame_slider_range_for_mode()
self._sync_frame_slider_with_current_frame()
self.update_image()
self._sync_frame_order_combo_with_mode()
return
if index == 1:
self._maxres_order_active = False
if not self.show_strong_peaks_checkbox.isEnabled():
self._sync_frame_order_combo_with_mode()
return
if not self.show_strong_peaks_checkbox.isChecked():
self.show_strong_peaks_checkbox.setChecked(True)
return
self._set_frame_slider_range_for_mode()
self._sync_frame_slider_with_current_frame()
self.update_image()
return
if index == 2:
self._activate_maxres_order_mode()
def _set_frame_slider_range_for_mode(self):
if self.total_frames <= 0:
self.frame_slider.setMaximum(0)
return
if self._is_show_maxres_active():
if not self._ensure_maxres_order(auto_generate=False):
self.frame_slider.setMaximum(max(self.total_frames - 1, 0))
return
self.frame_slider.setMaximum(max(len(self._maxres_order) - 1, 0))
elif self._is_show_strong_peaks_active():
if not self._ensure_strong_peaks_order():
self.frame_slider.setMaximum(max(self.total_frames - 1, 0))
return
self.frame_slider.setMaximum(max(len(self._strong_peaks_order) - 1, 0))
else:
self.frame_slider.setMaximum(max(self.total_frames - 1, 0))
def _sync_frame_slider_with_current_frame(self):
if self.total_frames <= 0:
return
if self._is_show_maxres_active():
if not self._ensure_maxres_order(auto_generate=False):
slider_value = int(self.current_frame_index)
else:
slider_value = self._maxres_rank.get(int(self.current_frame_index), 0)
elif self._is_show_strong_peaks_active():
if not self._ensure_strong_peaks_order():
slider_value = int(self.current_frame_index)
else:
slider_value = self._strong_peaks_rank.get(int(self.current_frame_index), 0)
else:
slider_value = int(self.current_frame_index)
slider_value = max(0, min(slider_value, self.frame_slider.maximum()))
self.frame_slider.blockSignals(True)
self.frame_slider.setValue(slider_value)
self.frame_slider.blockSignals(False)
[docs]
def on_show_strong_peaks_toggled(self, state):
# Use checkbox state directly to avoid enum/int signal mismatches.
enabled = bool(self.show_strong_peaks_checkbox.isChecked())
if enabled:
self._maxres_order_active = False
# Rebuild from current nPeaks data whenever strong-peaks mode is enabled.
self._clear_strong_peaks_order()
if enabled and not self._ensure_strong_peaks_order():
QMessageBox.warning(
self,
"nPeaks Order Unavailable",
"nPeaks dataset missing or unreadable. Run Find peaks first.",
)
self.show_strong_peaks_checkbox.blockSignals(True)
self.show_strong_peaks_checkbox.setChecked(False)
self.show_strong_peaks_checkbox.blockSignals(False)
self._sync_frame_order_combo_with_mode()
return
if enabled and self._strong_peaks_order:
# Entering strong-peaks mode should start from the strongest frame.
self.current_frame_index = int(self._strong_peaks_order[0])
self._set_frame_slider_range_for_mode()
self._sync_frame_slider_with_current_frame()
self._sync_frame_order_combo_with_mode()
self.update_image()
[docs]
def load_image_from_hdf5(self):
if not self.hdf5_path or not os.path.exists(self.hdf5_path):
log_print("Invalid HDF5 path:", self.hdf5_path)
return None, 0
else:
with h5py.File(self.hdf5_path, 'r') as file:
image_data = file['entry/data/images'][self.current_frame_index, :, :]
self._h5_height, self._h5_width = image_data.shape
image_shape = (self._h5_height, self._h5_width)
# Store raw grayscale intensities for the inspector
self.raw_image = image_data
# Safe normalization to avoid divide-by-zero
minv = image_data.min()
maxv = image_data.max()
denom = maxv - minv
if denom != 0:
image_norm = (image_data - minv) / denom
else:
image_norm = np.zeros_like(image_data)
image_norm = np.nan_to_num(image_norm, nan=0.0)
image_norm = np.clip(image_norm, 0.0, 1.0)
# Apply selected Matplotlib colormap, with optional log normalization
cmap = plt.get_cmap(self.cmap_combo.currentText())
if self.log_checkbox.isChecked():
# Use logarithmic normalization for high dynamic range
norm = colors.LogNorm(vmin=max(image_norm.min(), 1e-6), vmax=1.0)
mapped = cmap(norm(image_norm))
else:
mapped = cmap(image_norm)
rgb = (mapped[:, :, :3] * 255).astype(np.uint8)
self.original_image = Image.fromarray(rgb, 'RGB')
self._current_center_xy = None
draw = ImageDraw.Draw(self.original_image)
if 'entry/data/center_x' in file and 'entry/data/center_y' in file:
center_x = float(file['entry/data/center_x'][self.current_frame_index])
center_y = float(file['entry/data/center_y'][self.current_frame_index])
if np.isfinite(center_x) and np.isfinite(center_y):
self._current_center_xy = (center_x, center_y)
h5_peak_coords = []
if 'entry/data/peakXPosRaw' in file and 'entry/data/peakYPosRaw' in file:
peak_x_positions = file['entry/data/peakXPosRaw'][self.current_frame_index, :]
peak_y_positions = file['entry/data/peakYPosRaw'][self.current_frame_index, :]
max_peaks = min(len(peak_x_positions), len(peak_y_positions))
if 'entry/data/nPeaks' in file:
try:
declared = int(file['entry/data/nPeaks'][self.current_frame_index])
if declared < 0:
declared = 0
max_peaks = min(max_peaks, declared)
except Exception:
pass
peak_x_positions = peak_x_positions[:max_peaks]
peak_y_positions = peak_y_positions[:max_peaks]
for x, y in zip(peak_x_positions, peak_y_positions):
if np.isfinite(x) and np.isfinite(y) and x >= 0 and y >= 0:
h5_peak_coords.append((float(x), float(y)))
if self.show_peaks_checkbox.isChecked():
for x, y in h5_peak_coords:
draw.ellipse(
[x - 5, y - 5, x + 5, y + 5], outline='green', width=1
)
if (
getattr(self, 'show_indexing_results_action', None)
and self.show_indexing_results_action.isChecked()
):
chunk = self.indexing_manager.get_chunk_for_frame(self.current_frame_index)
if chunk:
reflections = self.indexing_manager.get_reflections_for_frame(
self.current_frame_index, chunk, image_shape
)
for x, y in reflections:
if x is None or y is None:
continue
draw.rectangle(
[x - 6, y - 6, x + 6, y + 6],
outline='cyan',
width=1
)
stream_peaks = self.indexing_manager.get_stream_peaks_for_frame(
self.current_frame_index, chunk, image_shape
)
self._warn_if_peak_mismatch(self.current_frame_index, h5_peak_coords, stream_peaks)
else:
self._indexing_debug(f"Frame {self.current_frame_index}: no stream chunk, skipping overlay.")
if getattr(self, 'show_resolution_rings_checkbox', None) and self.show_resolution_rings_checkbox.isChecked():
error = self._draw_resolution_rings(draw, file)
if error and not getattr(self, '_rings_warning_shown', False):
self._rings_warning_shown = True
QMessageBox.warning(self, "Resolution Rings", error)
n_peaks = 0 # Default value if nPeaks is not found
if 'entry/data/nPeaks' in file:
n_peaks = file['entry/data/nPeaks'][self.current_frame_index]
return self.original_image, n_peaks
"""def load_image_from_hdf5(self):
try:
with h5py.File(self.hdf5_path, 'r') as file:
# Load only a small section of the image for testing
small_image_data = file['entry/data/images'][self.current_frame_index, 0:256, 0:256]
log_print(f"Small image section loaded: {small_image_data.shape}")
image_data_normalized = (small_image_data - small_image_data.min()) / (
small_image_data.max() - small_image_data.min()
)
inferno_colormap = plt.colormaps['inferno']
colored_image = inferno_colormap(image_data_normalized)
colored_image = (colored_image[:, :, :3] * 255).astype(np.uint8)
self.original_image = Image.fromarray(colored_image, 'RGB')
return self.original_image, 0
except Exception as e:
log_print(f"Error in load_image_from_hdf5: {e}")
return None, 0"""
[docs]
def export_current_frame_as_shown(self):
"""Use helper to export the displayed (adjusted) image."""
pixmap = getattr(self, "_export_display_pixmap", None)
if pixmap is not None and not pixmap.isNull():
export_display_pixmap(self, pixmap)
else:
export_display_image(self, self.original_image)
[docs]
def export_current_frame(self):
"""Use helper to export the raw HDF5 frame as a TIFF."""
export_raw_image(self, self.raw_image)
[docs]
def adjust_gamma_value(self):
value = self.gamma_slider.value() / 10
self.gamma_label.setText(f"{value:.1f}")
if not self.is_slider_moving:
self.update_image()
[docs]
def adjust_brightness_value(self):
value = self.brightness_slider.value() / 10
self.brightness_label.setText(f"{value:.1f}")
if not self.is_slider_moving:
self.update_image()
[docs]
def adjust_contrast_value(self):
value = self.contrast_slider.value() / 10
self.contrast_label.setText(f"{value:.1f}")
if not self.is_slider_moving:
self.update_image()
[docs]
def on_slider_start(self):
self.is_slider_moving = True
[docs]
def on_slider_release(self):
if self.is_slider_moving:
self.is_slider_moving = False
self.update_image()
[docs]
def update_image(self):
try:
if self.hdf5_path is not None:
self.original_image, n_peaks = self.load_image_from_hdf5()
else:
return
# Update aspect ratio of the display box
width, height = self.original_image.size
self.image_aspect_widget.set_aspect_ratio(width / height if height else 1.0)
# Activate GUI elements
if self.current_frame_index == 0:
self.prev_button.setEnabled(False)
else:
self.prev_button.setEnabled(True)
if self.current_frame_index == self.total_frames - 1:
self.next_button.setEnabled(False)
else:
self.next_button.setEnabled(True)
if self.original_image is None:
log_print("No image to display.")
self.export_current_frame_action.setEnabled(False)
self.export_current_frame_as_displayed_action.setEnabled(False)
return
else:
self.export_current_frame_action.setEnabled(True)
self.export_current_frame_as_displayed_action.setEnabled(True)
# Get original image dimensions
original_width, original_height = self.original_image.size
# Apply gamma, brightness, and contrast corrections
adjusted_image = self.adjust_gamma(np.array(self.original_image), self.gamma_slider.value() / 10)
adjusted_image = Image.fromarray(adjusted_image)
adjusted_image = self.adjust_brightness(adjusted_image, self.brightness_slider.value() / 10)
adjusted_image = self.adjust_contrast(adjusted_image, self.contrast_slider.value() / 10)
geometry_applied = False
self._geometry_display_context = None
# Apply pixel mask after enhancements
try:
with h5py.File(self.hdf5_path, 'r') as mf:
pm_ds = get_mask_dataset(mf)
if pm_ds is not None and pm_ds.ndim == 2 and pm_ds.shape == (original_height, original_width):
mask_arr = pm_ds[()]
alpha = Image.fromarray((mask_arr.astype(np.uint8) * 255), 'L')
adjusted_image = adjusted_image.convert('RGBA')
adjusted_image.putalpha(alpha)
except Exception as e:
log_print(f"Error applying pixel mask: {e}")
# Apply live preview if active
if self.live_preview_active:
s = self.current_peak_settings
algo = s.get('algorithm', 'peakfinder8')
if algo == 'peakfinder9':
pf9_keys = [
'x0', 'y0', 'min_res', 'max_res',
'window_radius', 'min_sigma', 'min_peak_over_neighbors',
'min_snr_max_pixel', 'min_snr_peak_pixels', 'min_snr_whole_peak',
]
missing = [k for k in pf9_keys if k not in s]
if missing:
QMessageBox.warning(self, "Incomplete Settings",
f"Missing pf9 settings: {missing}")
else:
x0 = float(s['x0'])
y0 = float(s['y0'])
min_res = float(s['min_res'])
max_res = float(s['max_res'])
# Build radial mask (self.raw_image already loaded above)
X, Y = np.meshgrid(
range(self.raw_image.shape[1]),
range(self.raw_image.shape[0]),
)
R = np.sqrt((X - x0) ** 2 + (Y - y0) ** 2).astype(np.float32)
mask = np.ones_like(self.raw_image, dtype=np.int8)
mask[R > max_res] = 0
mask[R < min_res] = 0
try:
with h5py.File(self.hdf5_path, 'r') as mf:
pm_ds = get_mask_dataset(mf)
if (pm_ds is not None and pm_ds.ndim == 2
and pm_ds.shape == self.raw_image.shape):
mask = mask * pm_ds[()].astype(np.int8)
except Exception:
pass
xs, ys, _ = find_peaks_pf9_multiscale(
self.raw_image, mask,
window_radii = s['window_radius'],
min_sigma = float(s['min_sigma']),
min_peak_over_neighbors = float(s['min_peak_over_neighbors']),
min_snr_max_pixel = float(s['min_snr_max_pixel']),
min_snr_peak_pixels = float(s['min_snr_peak_pixels']),
min_snr_whole_peak = float(s['min_snr_whole_peak']),
)
peaks = list(zip(xs, ys))
adjusted_image = self.draw_peak_preview(adjusted_image, peaks)
n_peaks = len(peaks)
else: # peakfinder8
expected_keys = [
'x0', 'y0', 'threshold', 'min_snr',
'min_pix_count', 'max_pix_count', 'local_bg_radius',
'min_res', 'max_res',
]
filtered_settings = {}
for k in expected_keys:
if k in s:
try:
filtered_settings[k] = float(s[k])
except ValueError:
log_print(f"Warning: Unable to convert setting '{k}' to float.")
else:
log_print(f"Warning: Missing peak finder setting '{k}'.")
if len(filtered_settings) != len(expected_keys):
QMessageBox.warning(self, "Incomplete Settings",
"Some peak finder settings are missing or invalid.")
else:
peaks = process_single_frame(
self.hdf5_path, self.current_frame_index, **filtered_settings
)
adjusted_image = self.draw_peak_preview(adjusted_image, peaks)
n_peaks = len(peaks)
# Geometry remap is optional and only applied when [DetectorGeometry] exists.
try:
remapped = self._apply_detector_geometry_transform(adjusted_image)
geometry_applied = remapped is not adjusted_image
adjusted_image = remapped
except Exception as e:
log_print(f"Error applying detector geometry transform: {e}")
geometry_applied = False
# Convert PIL image to QPixmap using ImageQt
qt_image = ImageQt(adjusted_image)
pixmap = QPixmap.fromImage(qt_image)
# Automatic scaling: use container width, maintain original aspect ratio
orig_w, orig_h = pixmap.width(), pixmap.height()
ratio = orig_w / orig_h if orig_h else 1.0
self.image_aspect_widget.set_aspect_ratio(ratio)
container_width = self.image_aspect_widget.width()
new_height = int(container_width / ratio)
# Build full-resolution composited display image for export.
self._export_display_pixmap = pixmap.copy()
self._draw_display_overlays(
self._export_display_pixmap, geometry_applied, sx=1.0, sy=1.0
)
scaled_pixmap = self._export_display_pixmap.scaled(
container_width, new_height,
Qt.AspectRatioMode.KeepAspectRatio,
Qt.TransformationMode.SmoothTransformation
)
disp_w = scaled_pixmap.width()
disp_h = scaled_pixmap.height()
self.image_label.setPixmap(scaled_pixmap)
# orig_w, orig_h were set earlier from pixmap.width()/height()
self.x_scale_factor = orig_w / disp_w
self.y_scale_factor = orig_h / disp_h
# Adjust container height to match scaled pixmap
self.image_aspect_widget.setFixedHeight(new_height)
# Update the frame number in the entry box
self.frame_entry.setText(str(self.current_frame_index + 1))
# Update the info labels
self.frame_label.setText(f"Frame: {self.current_frame_index + 1}/{self.total_frames}")
self._sync_frame_slider_with_current_frame()
self.n_peaks_label.setText(f"Number of Peaks: {n_peaks}")
# Update beam center coordinates label
if self.hdf5_path:
with h5py.File(self.hdf5_path, 'r') as f:
if 'entry/data/center_x' in f and 'entry/data/center_y' in f:
cx = f['entry/data/center_x'][self.current_frame_index]
cy = f['entry/data/center_y'][self.current_frame_index]
self.beam_center_label.setText(f"Beam Center: {cx:.1f}, {cy:.1f}")
else:
self.beam_center_label.setText("Beam Center: N/A")
# Update slider labels to reflect current values
self.gamma_label.setText(f"{self.gamma_slider.value()/10:.1f}")
self.brightness_label.setText(f"{self.brightness_slider.value()/10:.1f}")
self.contrast_label.setText(f"{self.contrast_slider.value()/10:.1f}")
except Exception as e:
QMessageBox.critical(self, "Error", f"An error occurred while updating the image:\n{str(e)}")
[docs]
def adjust_gamma(self, image, gamma=1.0):
inv_gamma = 1.0 / gamma
table = np.array(
[((i / 255.0) ** inv_gamma) * 255 for i in np.arange(256)]
).astype("uint8")
return cv2.LUT(np.array(image, dtype=np.uint8), table)
[docs]
def adjust_brightness(self, image, brightness_factor):
return ImageEnhance.Brightness(image).enhance(brightness_factor)
[docs]
def adjust_contrast(self, image, contrast_factor):
return ImageEnhance.Contrast(image).enhance(contrast_factor)
[docs]
def change_frame_from_entry(self):
try:
frame_num = int(self.frame_entry.text()) - 1
if 0 <= frame_num < self.total_frames:
self.current_frame_index = frame_num
self.update_image()
else:
QMessageBox.warning(self, "Warning", "Frame number out of range.")
except ValueError:
QMessageBox.warning(self, "Warning", "Invalid input. Please enter a number.")
[docs]
def change_frame(self, value):
"""
Slot to handle changes in the frame_slider's value.
Parameters:
value (int): The new value from the slider representing the frame index.
"""
try:
# Ignore slider events while no data is loaded
if self.total_frames <= 0:
return
# Convert the slider value to an integer frame index (or strong-peaks rank index)
frame_num = round(float(value))
if self._is_show_maxres_active():
if not self._ensure_maxres_order(auto_generate=False):
return
if 0 <= frame_num < len(self._maxres_order):
self.current_frame_index = int(self._maxres_order[frame_num])
self.update_image()
else:
QMessageBox.warning(self, "Warning", f"Frame rank {frame_num} is out of range.")
elif self._is_show_strong_peaks_active():
if not self._ensure_strong_peaks_order():
return
if 0 <= frame_num < len(self._strong_peaks_order):
self.current_frame_index = int(self._strong_peaks_order[frame_num])
self.update_image()
else:
QMessageBox.warning(self, "Warning", f"Frame rank {frame_num} is out of range.")
else:
# Ensure the frame number is within valid range
if 0 <= frame_num < self.total_frames:
self.current_frame_index = frame_num
self.update_image()
else:
QMessageBox.warning(self, "Warning", f"Frame index {frame_num} is out of range.")
except ValueError:
QMessageBox.warning(self, "Warning", "Invalid frame value received from slider.")
def _ini_acquisition_details(self):
"""Return the AcquisitionDetails section of the current INI as a dict, or {}."""
ini = getattr(self, 'full_ini_file_path', None)
if not ini:
return {}
try:
cfg = configparser.ConfigParser(
interpolation=None,
inline_comment_prefixes=(";", "#"),
)
cfg.read(ini)
if cfg.has_section('AcquisitionDetails'):
return dict(cfg.items('AcquisitionDetails'))
except Exception:
pass
return {}
def _parse_pair(self, value, cast=float):
if value is None:
return None
parts = [p.strip() for p in str(value).split(",")]
if len(parts) != 2:
return None
try:
return cast(parts[0]), cast(parts[1])
except Exception:
return None
def _parse_matrix_2x2(self, value):
if value is None:
return None
parts = [p.strip() for p in str(value).split(",")]
if len(parts) != 4:
return None
try:
a11, a12, a21, a22 = (float(p) for p in parts)
return np.array([[a11, a12], [a21, a22]], dtype=np.float64)
except Exception:
return None
def _load_detector_geometry_from_ini(self):
"""
Load optional detector geometry from [DetectorGeometry].
Returns None when geometry is absent/disabled/invalid.
"""
ini_path = getattr(self, 'full_ini_file_path', None)
if not ini_path or not os.path.exists(ini_path):
return None
cache_key = (os.path.abspath(ini_path), os.path.getmtime(ini_path))
if getattr(self, "_det_geom_cache_key", None) == cache_key:
return getattr(self, "_det_geom_cache_data", None)
geometry = None
try:
cfg = configparser.ConfigParser(
interpolation=None,
inline_comment_prefixes=(";", "#"),
)
cfg.read(ini_path)
section = "DetectorGeometry"
if not cfg.has_section(section):
self._det_geom_cache_key = cache_key
self._det_geom_cache_data = None
return None
sec = cfg[section]
enabled_raw = str(sec.get("enabled", "true")).strip().lower()
if enabled_raw in {"0", "false", "no", "off"}:
self._det_geom_cache_key = cache_key
self._det_geom_cache_data = None
return None
global_offset = self._parse_pair(sec.get("global_offset", "0,0"), float)
if global_offset is None:
global_offset = (0.0, 0.0)
global_A = self._parse_matrix_2x2(sec.get("global_a", "1,0,0,1"))
if global_A is None:
global_A = np.eye(2, dtype=np.float64)
raw_x_direction = str(sec.get("raw_x_direction", "+right")).strip().lower()
raw_y_direction = str(sec.get("raw_y_direction", "+up")).strip().lower()
x_sign = 1 if raw_x_direction in {"+right", "right", "+x", "x"} else -1
y_sign = 1 if raw_y_direction in {"+down", "down", "+y", "y"} else -1
panel_ids_raw = str(sec.get("panels", "")).strip()
if panel_ids_raw:
panel_ids = [p.strip().lower() for p in panel_ids_raw.split(",") if p.strip()]
else:
panel_ids = []
for key in sec.keys():
if key.endswith("_raw_min"):
panel_ids.append(key[:-8].strip().lower())
panel_ids = sorted(set(panel_ids))
panels = []
for panel_id in panel_ids:
raw_min = self._parse_pair(sec.get(f"{panel_id}_raw_min"), float)
raw_max = self._parse_pair(sec.get(f"{panel_id}_raw_max"), float)
offset = self._parse_pair(sec.get(f"{panel_id}_offset", "0,0"), float)
matrix = self._parse_matrix_2x2(sec.get(f"{panel_id}_a", "1,0,0,1"))
if raw_min is None or raw_max is None or offset is None or matrix is None:
continue
if np.allclose(matrix, 0.0):
# User-friendly fallback: interpret all-zero matrix as "no rotation".
matrix = np.eye(2, dtype=np.float64)
det = float(np.linalg.det(matrix))
if abs(det) < 1e-10:
log_print(f"Skipping panel '{panel_id}' due to singular transform matrix.")
continue
raw_min_x, raw_min_y = raw_min
raw_max_x, raw_max_y = raw_max
if raw_min_x > raw_max_x or raw_min_y > raw_max_y:
continue
panels.append(
{
"id": panel_id,
"raw_min_x": float(raw_min_x),
"raw_min_y": float(raw_min_y),
"raw_max_x": float(raw_max_x),
"raw_max_y": float(raw_max_y),
"offset": np.array(offset, dtype=np.float64),
"A": matrix,
}
)
if panels:
geometry = {
"pixel_convention": str(sec.get("pixel_convention", "center")).strip().lower(),
"x_sign": x_sign,
"y_sign": y_sign,
"global_offset": np.array(global_offset, dtype=np.float64),
"global_A": global_A,
"panels": panels,
}
except Exception as exc:
log_print(f"Failed to load [DetectorGeometry] from INI: {exc}")
geometry = None
self._det_geom_cache_key = cache_key
self._det_geom_cache_data = geometry
return geometry
def _img_to_raw_coords(self, x_img, y_img, width, height, x_sign, y_sign):
x_bias = 0.0 if x_sign == 1 else float(width - 1)
y_bias = 0.0 if y_sign == 1 else float(height - 1)
raw_x = x_sign * float(x_img) + x_bias
raw_y = y_sign * float(y_img) + y_bias
return raw_x, raw_y
def _raw_to_img_coords(self, raw_x, raw_y, width, height, x_sign, y_sign):
x_bias = 0.0 if x_sign == 1 else float(width - 1)
y_bias = 0.0 if y_sign == 1 else float(height - 1)
x_img = (float(raw_x) - x_bias) / x_sign
y_img = (float(raw_y) - y_bias) / y_sign
return x_img, y_img
def _panel_contains_raw(self, panel, raw_x, raw_y):
return (
panel["raw_min_x"] <= raw_x <= panel["raw_max_x"]
and panel["raw_min_y"] <= raw_y <= panel["raw_max_y"]
)
def _raw_to_geometry_canvas(self, raw_x, raw_y, panel, context):
raw_min = np.array([panel["raw_min_x"], panel["raw_min_y"]], dtype=np.float64)
delta = np.array(
[float(raw_x) - panel["raw_min_x"], float(raw_y) - panel["raw_min_y"]],
dtype=np.float64,
)
# Offsets are defined as shifts relative to the panel's raw position.
corrected = raw_min + panel["offset"] + panel["A"] @ delta
global_xy = context["global_offset"] + context["global_A"] @ corrected
x_canvas = float(global_xy[0] + context["shift_x"])
y_canvas = float(global_xy[1] + context["shift_y"])
return x_canvas, y_canvas
def _image_to_geometry_canvas(self, x_img, y_img, context):
"""Map source-image coordinates to geometry-canvas coordinates."""
if not context:
return None
raw_x, raw_y = self._img_to_raw_coords(
x_img, y_img,
context["width"], context["height"],
context["x_sign"], context["y_sign"],
)
for panel in context.get("panels", []):
if self._panel_contains_raw(panel, raw_x, raw_y):
return self._raw_to_geometry_canvas(raw_x, raw_y, panel, context)
# Gap fallback: apply only global transform so detector-level coordinates
# (e.g. beam center in a chip gap) remain visible.
global_xy = context["global_offset"] + (context["global_A"] @ np.array([raw_x, raw_y], dtype=np.float64))
return float(global_xy[0] + context["shift_x"]), float(global_xy[1] + context["shift_y"])
def _build_detector_geometry_context(self, image_shape, geometry):
if geometry is None:
return None
height, width = image_shape
if height <= 0 or width <= 0:
return None
x_sign = int(geometry["x_sign"])
y_sign = int(geometry["y_sign"])
prepared_panels = []
min_x = np.inf
min_y = np.inf
max_x = -np.inf
max_y = -np.inf
for panel in geometry["panels"]:
c1, r1 = self._raw_to_img_coords(
panel["raw_min_x"], panel["raw_min_y"], width, height, x_sign, y_sign
)
c2, r2 = self._raw_to_img_coords(
panel["raw_max_x"], panel["raw_max_y"], width, height, x_sign, y_sign
)
col_min = int(np.floor(min(c1, c2)))
col_max = int(np.ceil(max(c1, c2)))
row_min = int(np.floor(min(r1, r2)))
row_max = int(np.ceil(max(r1, r2)))
col_min = max(0, min(width - 1, col_min))
col_max = max(0, min(width - 1, col_max))
row_min = max(0, min(height - 1, row_min))
row_max = max(0, min(height - 1, row_max))
if col_min > col_max or row_min > row_max:
continue
roi_w = col_max - col_min + 1
roi_h = row_max - row_min + 1
def local_to_global(lx, ly):
x_img = col_min + float(lx)
y_img = row_min + float(ly)
raw_x, raw_y = self._img_to_raw_coords(x_img, y_img, width, height, x_sign, y_sign)
raw_min = np.array([panel["raw_min_x"], panel["raw_min_y"]], dtype=np.float64)
delta = np.array(
[raw_x - panel["raw_min_x"], raw_y - panel["raw_min_y"]],
dtype=np.float64,
)
corrected = raw_min + panel["offset"] + panel["A"] @ delta
global_xy = geometry["global_offset"] + geometry["global_A"] @ corrected
return float(global_xy[0]), float(global_xy[1])
g00 = local_to_global(0.0, 0.0)
g10 = local_to_global(1.0, 0.0)
g01 = local_to_global(0.0, 1.0)
affine_base = np.array(
[
[g10[0] - g00[0], g01[0] - g00[0], g00[0]],
[g10[1] - g00[1], g01[1] - g00[1], g00[1]],
],
dtype=np.float64,
)
corners = [
local_to_global(0.0, 0.0),
local_to_global(float(roi_w), 0.0),
local_to_global(0.0, float(roi_h)),
local_to_global(float(roi_w), float(roi_h)),
]
xs = [p[0] for p in corners]
ys = [p[1] for p in corners]
min_x = min(min_x, min(xs))
min_y = min(min_y, min(ys))
max_x = max(max_x, max(xs))
max_y = max(max_y, max(ys))
prepared_panels.append(
{
"id": panel["id"],
"raw_min_x": panel["raw_min_x"],
"raw_min_y": panel["raw_min_y"],
"raw_max_x": panel["raw_max_x"],
"raw_max_y": panel["raw_max_y"],
"offset": panel["offset"],
"A": panel["A"],
"col_min": col_min,
"col_max": col_max,
"row_min": row_min,
"row_max": row_max,
"affine_base": affine_base,
}
)
if not prepared_panels or not np.isfinite(min_x + min_y + max_x + max_y):
return None
out_w = int(np.ceil(max_x - min_x))
out_h = int(np.ceil(max_y - min_y))
if out_w <= 0 or out_h <= 0:
return None
if out_w > 20000 or out_h > 20000:
log_print(f"Skipping detector geometry transform: target canvas too large ({out_w}x{out_h})")
return None
shift_x = -float(min_x)
shift_y = -float(min_y)
for panel in prepared_panels:
affine = panel["affine_base"].copy()
affine[0, 2] += shift_x
affine[1, 2] += shift_y
panel["affine"] = affine
linear = affine[:, :2]
try:
panel["affine_inv"] = np.linalg.inv(linear)
except np.linalg.LinAlgError:
panel["affine_inv"] = None
return {
"width": width,
"height": height,
"x_sign": x_sign,
"y_sign": y_sign,
"global_offset": geometry["global_offset"],
"global_A": geometry["global_A"],
"panels": prepared_panels,
"shift_x": shift_x,
"shift_y": shift_y,
"out_w": out_w,
"out_h": out_h,
}
def _apply_detector_geometry_transform(self, image):
"""
Apply optional detector geometry remap to the display image.
Pixel masking should be applied first; geometry remap then moves pixels
including their alpha values.
"""
self._geometry_display_context = None
if image is None:
return image
geometry = self._load_detector_geometry_from_ini()
if geometry is None:
return image
src_rgba = np.asarray(image.convert("RGBA"))
context = self._build_detector_geometry_context(src_rgba.shape[:2], geometry)
if context is None:
return image
canvas = np.zeros((context["out_h"], context["out_w"], 4), dtype=np.float32)
for panel in context["panels"]:
roi = src_rgba[
panel["row_min"]:panel["row_max"] + 1,
panel["col_min"]:panel["col_max"] + 1,
]
warped = cv2.warpAffine(
roi,
panel["affine"].astype(np.float32),
(context["out_w"], context["out_h"]),
flags=cv2.INTER_LINEAR,
borderMode=cv2.BORDER_CONSTANT,
borderValue=(0, 0, 0, 0),
)
warped_rgb = warped[..., :3].astype(np.float32)
warped_alpha = warped[..., 3:4].astype(np.float32) / 255.0
inv_alpha = 1.0 - warped_alpha
canvas[..., :3] = warped_rgb * warped_alpha + canvas[..., :3] * inv_alpha
canvas[..., 3:4] = 255.0 * (warped_alpha + (canvas[..., 3:4] / 255.0) * inv_alpha)
self._geometry_display_context = context
output = np.clip(canvas, 0, 255).astype(np.uint8)
return Image.fromarray(output, "RGBA")
def _geometry_canvas_to_image_coords(self, x_canvas, y_canvas, context):
"""
Map a click in geometry-remapped canvas coordinates back to original image coords.
Returns (x_img, y_img) in source image coordinates (top-left origin), or None.
"""
px = float(x_canvas)
py = float(y_canvas)
for panel in context.get("panels", []):
affine = panel.get("affine")
inv = panel.get("affine_inv")
if affine is None or inv is None:
continue
t = affine[:, 2]
local = inv @ (np.array([px, py], dtype=np.float64) - t)
lx = float(local[0])
ly = float(local[1])
roi_w = panel["col_max"] - panel["col_min"] + 1
roi_h = panel["row_max"] - panel["row_min"] + 1
# Small tolerance accounts for interpolation at panel borders.
if not (-0.5 <= lx <= roi_w - 0.5 and -0.5 <= ly <= roi_h - 0.5):
continue
x_img = panel["col_min"] + lx
y_img = panel["row_min"] + ly
return x_img, y_img
return None
def _draw_detector_geometry_resolution_rings(self, painter, rings, cx_img, cy_img, sx, sy):
"""Draw rings in geometry-remapped view across all mapped detector panels."""
context = getattr(self, "_geometry_display_context", None)
if not context:
return False
panels = context.get("panels", [])
if not panels:
return False
def panel_for_raw(raw_x, raw_y):
for panel in panels:
if self._panel_contains_raw(panel, raw_x, raw_y):
return panel
return None
for r_x, r_y, d_A in rings:
path = QPainterPath()
started = False
label_pos = None
last_panel_id = None
steps = 240
for i in range(steps + 1):
theta = (2.0 * np.pi * i) / steps
x_img = cx_img + r_x * np.cos(theta)
y_img = cy_img + r_y * np.sin(theta)
raw_x, raw_y = self._img_to_raw_coords(
x_img, y_img,
context["width"], context["height"],
context["x_sign"], context["y_sign"],
)
panel = panel_for_raw(raw_x, raw_y)
if panel is None:
if started:
painter.drawPath(path)
path = QPainterPath()
started = False
last_panel_id = None
continue
panel_id = panel.get("id")
if started and panel_id != last_panel_id:
painter.drawPath(path)
path = QPainterPath()
started = False
x_canvas, y_canvas = self._raw_to_geometry_canvas(raw_x, raw_y, panel, context)
dx = x_canvas * sx
dy = y_canvas * sy
if label_pos is None:
label_pos = (dx, dy)
if not started:
path.moveTo(dx, dy)
started = True
else:
path.lineTo(dx, dy)
last_panel_id = panel_id
if started:
painter.drawPath(path)
if label_pos is not None:
painter.drawText(QPointF(label_pos[0] + 2, label_pos[1] + 4), f"{d_A:.1f}A")
return True
def _draw_display_overlays(self, pixmap, geometry_applied: bool, sx: float, sy: float):
"""Draw rings and beam-center marker on a pixmap in display coordinates."""
draw_rings = (
getattr(self, 'show_resolution_rings_checkbox', None)
and self.show_resolution_rings_checkbox.isChecked()
and getattr(self, '_pending_rings', None)
)
draw_center = bool(
self.show_point_checkbox.isChecked()
and isinstance(getattr(self, '_current_center_xy', None), tuple)
)
vdf_overlay = getattr(self, "_vdf_overlay_settings", None)
draw_vdf = bool(
vdf_overlay
and vdf_overlay.get("enabled")
and isinstance(getattr(self, '_current_center_xy', None), tuple)
)
if not (draw_rings or draw_center or draw_vdf):
return
painter = QPainter(pixmap)
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
if draw_vdf:
self._draw_vdf_overlay(painter, geometry_applied, sx, sy, vdf_overlay)
if draw_rings:
cx0, cy0, rings = self._pending_rings
ring_color = QColor(255, 160, 80)
pen = QPen(ring_color)
pen.setWidthF(0.0) # cosmetic pen: always 1 physical screen pixel
font = QFont()
font.setPointSize(7)
painter.setPen(pen)
painter.setFont(font)
if geometry_applied:
drawn = self._draw_detector_geometry_resolution_rings(
painter, rings, cx0, cy0, sx, sy
)
if not drawn:
for r_x, r_y, d_A in rings:
dx, dy = cx0 * sx, cy0 * sy
drx, dry = r_x * sx, r_y * sy
painter.drawEllipse(QRectF(dx - drx, dy - dry, 2 * drx, 2 * dry))
painter.drawText(QPointF(dx + drx + 2, dy + 4), f"{d_A:.1f}A")
else:
for r_x, r_y, d_A in rings:
dx, dy = cx0 * sx, cy0 * sy
drx, dry = r_x * sx, r_y * sy
painter.drawEllipse(QRectF(dx - drx, dy - dry, 2 * drx, 2 * dry))
painter.drawText(QPointF(dx + drx + 2, dy + 4), f"{d_A:.1f}A")
if draw_center:
cx_img, cy_img = self._current_center_xy
if geometry_applied:
context = getattr(self, "_geometry_display_context", None)
mapped = self._image_to_geometry_canvas(cx_img, cy_img, context)
if mapped is not None:
cx_disp = mapped[0] * sx
cy_disp = mapped[1] * sy
else:
cx_disp = cx_img * sx
cy_disp = cy_img * sy
else:
cx_disp = cx_img * sx
cy_disp = cy_img * sy
center_pen = QPen(QColor(255, 48, 48))
center_pen.setWidthF(0.0)
painter.setPen(center_pen)
crosshair_size = 10.0
painter.drawLine(
QPointF(cx_disp - crosshair_size, cy_disp),
QPointF(cx_disp + crosshair_size, cy_disp),
)
painter.drawLine(
QPointF(cx_disp, cy_disp - crosshair_size),
QPointF(cx_disp, cy_disp + crosshair_size),
)
painter.end()
def _draw_vdf_overlay(self, painter, geometry_applied: bool, sx: float, sy: float, settings):
"""Draw the virtual dark-field annulus on the current diffraction preview."""
cx_img, cy_img = self._current_center_xy
if geometry_applied:
context = getattr(self, "_geometry_display_context", None)
mapped = self._image_to_geometry_canvas(cx_img, cy_img, context)
if mapped is not None:
cx_disp = mapped[0] * sx
cy_disp = mapped[1] * sy
else:
cx_disp = cx_img * sx
cy_disp = cy_img * sy
else:
cx_disp = cx_img * sx
cy_disp = cy_img * sy
center = float(settings.get("center", 0.0))
width = max(1.0, float(settings.get("width", 1.0)))
inner = max(0.0, center - width / 2.0)
outer = max(inner + 1.0, center + width / 2.0)
path = QPainterPath()
path.setFillRule(Qt.FillRule.OddEvenFill)
path.addEllipse(QRectF(
cx_disp - outer * sx,
cy_disp - outer * sy,
2.0 * outer * sx,
2.0 * outer * sy,
))
if inner > 0:
path.addEllipse(QRectF(
cx_disp - inner * sx,
cy_disp - inner * sy,
2.0 * inner * sx,
2.0 * inner * sy,
))
painter.fillPath(path, QColor(60, 190, 255, 85))
pen = QPen(QColor(60, 220, 255, 220))
pen.setWidthF(0.0)
painter.setPen(pen)
painter.drawEllipse(QRectF(
cx_disp - outer * sx,
cy_disp - outer * sy,
2.0 * outer * sx,
2.0 * outer * sy,
))
if inner > 0:
painter.drawEllipse(QRectF(
cx_disp - inner * sx,
cy_disp - inner * sy,
2.0 * inner * sx,
2.0 * inner * sy,
))
[docs]
def set_vdf_overlay(self, center, width, enabled=True):
"""Set the virtual dark-field overlay shown on the main diffraction preview."""
self._vdf_overlay_settings = {
"center": int(center),
"width": int(width),
"enabled": bool(enabled),
}
if getattr(self, "hdf5_path", None):
self.update_image()
[docs]
def clear_vdf_overlay(self):
"""Remove the virtual dark-field detector overlay from the main preview."""
if getattr(self, "_vdf_overlay_settings", None) is None:
return
self._vdf_overlay_settings = None
if getattr(self, "hdf5_path", None):
self.update_image()
def _draw_resolution_rings(self, draw, file):
"""Draw resolution rings as circles/ellipses on the diffraction pattern image.
Falls back to the INI AcquisitionDetails when values are absent from HDF5.
Returns None on success, or an error string describing what metadata is missing.
"""
self._pending_rings = None
try:
missing = []
ini = self._ini_acquisition_details()
# --- beam center ---
if 'entry/data/center_x' not in file or 'entry/data/center_y' not in file:
missing.append("beam center (run 'Find Centers' first)")
else:
cx = float(file['entry/data/center_x'][self.current_frame_index])
cy = float(file['entry/data/center_y'][self.current_frame_index])
if np.isnan(cx) or np.isnan(cy):
missing.append("beam center for this frame (NaN — center finding may have failed here)")
# --- pixel size (m) ---
# HDF5 stores meters_per_pixel = 1 / pixels_per_meter
ps_x, ps_y = None, None
ps_x_ds = file.get('/entry/instrument/detector/x_pixel_size')
if ps_x_ds is not None:
ps_x = float(ps_x_ds[()])
ps_y_ds = file.get('/entry/instrument/detector/y_pixel_size')
ps_y = float(ps_y_ds[()]) if ps_y_ds is not None else ps_x
else:
# Fall back to pixels_per_meter from INI
ppm_raw = ini.get('pixels_per_meter', '')
try:
ppm = float(ppm_raw)
if ppm > 0:
ps_x = ps_y = 1.0 / ppm
except (ValueError, ZeroDivisionError):
pass
if ps_x is None:
missing.append("pixel size (set in Preflight Check)")
# --- effective camera length (m) ---
cam_len = None
cam_eff_ds = file.get('/entry/instrument/detector/camera_length_effective')
if cam_eff_ds is not None:
cam_len = float(cam_eff_ds[()])
else:
cam_raw_ds = file.get('/entry/instrument/detector/camera_length')
if cam_raw_ds is not None:
cam_len = float(cam_raw_ds[()])
corr_ds = file.get('/entry/instrument/detector/camera_length_correction')
if corr_ds is not None:
correction = float(corr_ds[()])
if correction > 0:
cam_len *= correction
else:
# Fall back to INI
cl_raw = ini.get('camera_length', '')
try:
cl = float(cl_raw)
if cl > 0:
corr = float(ini.get('camera_length_correction', '1') or '1')
cam_len = cl * (corr if corr > 0 else 1.0)
except ValueError:
pass
if cam_len is None:
missing.append("camera length (set in Preflight Check)")
# --- incident energy (eV) ---
# HDF5 stores incident_energy in eV (= acceleration_voltage in V)
energy_ev = None
energy_ds = file.get('/entry/instrument/beam/incident_energy')
if energy_ds is not None:
energy_ev = float(energy_ds[()])
else:
# Fall back to INI: acceleration_voltage is stored in volts
av_raw = ini.get('acceleration_voltage', '')
try:
energy_ev = float(av_raw) # volts == eV for electrons
except ValueError:
pass
if energy_ev is None:
missing.append("acceleration voltage / incident energy (set in Preflight Check)")
if missing:
return (
"Cannot draw resolution rings. The following metadata is missing:\n\n"
+ "\n".join(f" \u2022 {m}" for m in missing)
)
if cam_len <= 0 or energy_ev <= 0 or ps_x <= 0:
return "Cannot draw resolution rings: invalid (zero or negative) metadata values."
# --- relativistic electron wavelength (m) ---
h = 6.62607015e-34 # J·s
m_e = 9.1093837015e-31 # kg
e = 1.602176634e-19 # C
c = 299792458.0 # m/s
E_j = energy_ev * e
lam = h / np.sqrt(2 * m_e * E_j + E_j ** 2 / c ** 2)
# --- collect ring data (drawn later on the scaled pixmap via QPainter) ---
d_spacings_A = [3.0, 2.0, 1.5, 1.0, 0.8, 0.6]
rings = []
for d_A in d_spacings_A:
d_m = d_A * 1e-10
r_x = lam * cam_len / (d_m * ps_x)
r_y = lam * cam_len / (d_m * ps_y)
if not (np.isfinite(r_x) and np.isfinite(r_y)) or r_x < 3 or r_y < 3:
continue
rings.append((r_x, r_y, d_A))
self._pending_rings = (cx, cy, rings)
return None # success
except Exception as exc:
log_print(f"Error drawing resolution rings: {exc}", level="debug")
return f"Unexpected error drawing resolution rings: {exc}"
[docs]
def draw_peak_preview(self, image, peaks):
draw = ImageDraw.Draw(image)
for x, y in peaks:
draw.ellipse([x - 5, y - 5, x + 5, y + 5], outline='green', width=1)
return image
[docs]
def show_pixel_index(self, event):
# Get current pixmap
pixmap = self.image_label.pixmap()
if pixmap is None:
return
# Click position relative to the label
pos = event.position()
x = pos.x()
y = pos.y()
# Calculate pixmap offsets inside the label (centered)
label_w = self.image_label.width()
label_h = self.image_label.height()
pm_w = pixmap.width()
pm_h = pixmap.height()
offset_x = (label_w - pm_w) / 2
offset_y = (label_h - pm_h) / 2
# Coordinates relative to the pixmap
x_rel = x - offset_x
y_rel = y - offset_y
def resolve_clicked_indices():
if not (0 <= x_rel < pm_w and 0 <= y_rel < pm_h):
return None
x_canvas = float(x_rel * self.x_scale_factor)
y_canvas = float(y_rel * self.y_scale_factor)
context = getattr(self, "_geometry_display_context", None)
if context:
mapped = self._geometry_canvas_to_image_coords(x_canvas, y_canvas, context)
if mapped is None:
return None
x_img, y_img = mapped
else:
x_img, y_img = x_canvas, y_canvas
x_index = int(np.floor(x_img + 0.5))
y_index = int(np.floor(y_img + 0.5))
raw_image = getattr(self, "raw_image", None)
if raw_image is not None and hasattr(raw_image, "shape") and len(raw_image.shape) >= 2:
h, w = raw_image.shape[:2]
if not (0 <= x_index < w and 0 <= y_index < h):
return None
return x_index, y_index
# Check which mouse button and mode
if self.coordinate_selection_active:
# Only handle left-click in coordinate selection mode
if event.button() != Qt.MouseButton.LeftButton:
return
indices = resolve_clicked_indices()
if indices is not None:
x_index, y_index = indices
# Route coordinate to whichever dialog requested it
target = getattr(self, "coordinate_selection_target", None)
if target and hasattr(target, "receive_coordinate"):
target.receive_coordinate(x_index, y_index)
elif hasattr(self, "settings_window_instance") and hasattr(self.settings_window_instance, "receive_coordinate"):
self.settings_window_instance.receive_coordinate(x_index, y_index)
else:
QMessageBox.warning(self, "Warning", "Clicked outside the image boundaries.")
# Reset coordinate selection mode
self.coordinate_selection_active = False
self.image_label.setCursor(Qt.CursorShape.ArrowCursor)
self.coordinate_selection_target = None
if hasattr(self, "_clear_status_message"):
self._clear_status_message()
else:
# Only handle right-click for pixel info
if event.button() != Qt.MouseButton.RightButton:
return
indices = resolve_clicked_indices()
if indices is not None:
x_index, y_index = indices
# Load raw frame data
if not self.hdf5_path or not os.path.exists(self.hdf5_path):
QMessageBox.critical(self, "Error", "No valid HDF5 file loaded.")
return
with h5py.File(self.hdf5_path, 'r') as file:
frame_data = file['entry/data/images'][self.current_frame_index]
# Ensure indices are within the array bounds
if not (0 <= y_index < frame_data.shape[0] and 0 <= x_index < frame_data.shape[1]):
QMessageBox.warning(self, "Warning", "Clicked outside the image boundaries.")
return
# Get intensity and relative intensity
intensity_value = frame_data[int(y_index), int(x_index)]
min_intensity = frame_data.min()
max_intensity = frame_data.max()
if max_intensity != min_intensity:
relative_intensity = (intensity_value - min_intensity) / (max_intensity - min_intensity)
else:
relative_intensity = 0
QMessageBox.information(
self,
"Pixel Info",
f"Point in frame {self.current_frame_index}\n\n"
f"x-coordinate: {x_index}\n"
f"y-coordinate: {y_index}\n"
f"Intensity: {intensity_value}\n"
f"Relative Intensity: {relative_intensity:.2f}",
)
else:
QMessageBox.warning(self, "Warning", "Clicked outside the image boundaries.")
[docs]
def next_frame(self):
if self._is_show_maxres_active():
if not self._ensure_maxres_order(auto_generate=False):
return
pos = self._maxres_rank.get(int(self.current_frame_index), 0)
if pos < len(self._maxres_order) - 1:
self.current_frame_index = int(self._maxres_order[pos + 1])
self.update_image()
elif self._is_show_strong_peaks_active():
if not self._ensure_strong_peaks_order():
return
pos = self._strong_peaks_rank.get(int(self.current_frame_index), 0)
if pos < len(self._strong_peaks_order) - 1:
self.current_frame_index = int(self._strong_peaks_order[pos + 1])
self.update_image()
else:
if self.current_frame_index < self.total_frames - 1:
self.current_frame_index += 1
self.update_image()
[docs]
def previous_frame(self):
if self._is_show_maxres_active():
if not self._ensure_maxres_order(auto_generate=False):
return
pos = self._maxres_rank.get(int(self.current_frame_index), 0)
if pos > 0:
self.current_frame_index = int(self._maxres_order[pos - 1])
self.update_image()
elif self._is_show_strong_peaks_active():
if not self._ensure_strong_peaks_order():
return
pos = self._strong_peaks_rank.get(int(self.current_frame_index), 0)
if pos > 0:
self.current_frame_index = int(self._strong_peaks_order[pos - 1])
self.update_image()
else:
if self.current_frame_index > 0:
self.current_frame_index -= 1
self.update_image()
[docs]
def open_map(self):
if not self.hdf5_path:
QMessageBox.warning(self, "No HDF5 file", "Please open an HDF5 file first.")
return
# Open the MapWindow with the selected HDF5 path
self.map_window = MapWindow(self, self.hdf5_path, ini_path=getattr(self, "full_ini_file_path", None))
# Connect the signal emitted by MapWindow to handle the frame update
self.map_window.point_clicked_signal.connect(self.update_frame_from_map_click)
# Show the MapWindow as a dialog
self.map_window.show()
# The map window writes/refreshes the atlas on open; reflect it in the preview.
self.update_map_thumbnail()
def _default_atlas_zdim(self, h5_path):
"""Return the preferred full-frame intensity dataset for atlas generation."""
with h5py.File(h5_path, "r") as f:
data_group = f.get("entry/data")
if data_group is None:
return None
for name in ("frame_mean_intensities", "frame_total_intensities"):
if name in data_group:
return name
return None
def _atlas_exists(self, h5_path, zdim):
"""Return True when an atlas for ``zdim`` is already saved."""
if not zdim:
return False
with h5py.File(h5_path, "r") as f:
return f"entry/atlas/{zdim}/raster" in f
def _any_atlas_exists(self, h5_path):
"""Return True when any saved atlas exists in the file."""
with h5py.File(h5_path, "r") as f:
return "entry/atlas" in f and bool(f["entry/atlas"].keys())
def _atlas_backlash_offset_um(self):
"""Read the persistent atlas backlash offset from the selected INI."""
ini_path = getattr(self, "full_ini_file_path", None)
if not ini_path or not os.path.exists(ini_path):
return 0.0
config = configparser.ConfigParser()
try:
config.read(ini_path)
return config.getfloat("Parameters", "atlas_backlash_offset_um", fallback=0.0)
except Exception as exc:
log_print(f"Failed to read atlas backlash offset from INI: {exc}", level="warning")
return 0.0
[docs]
def ensure_atlas_for_current_file(self):
"""Create the default saved atlas when all inputs exist and no atlas is saved."""
if not self.hdf5_path or not os.path.exists(self.hdf5_path):
return False
try:
if self._any_atlas_exists(self.hdf5_path):
return False
zdim = self._default_atlas_zdim(self.hdf5_path)
if zdim is None or self._atlas_exists(self.hdf5_path, zdim):
return False
from coseda.map.streaks import assign_streaks, save_atlas
with h5py.File(self.hdf5_path, "r") as f:
data_group = f.get("entry/data")
if data_group is None:
return False
has_stage = (
("stagepos_x_refined" in data_group or "stagepos_x" in data_group)
and ("stagepos_y_refined" in data_group or "stagepos_y" in data_group)
)
if not has_stage:
return False
self._set_status_message("Reconstructing atlas", animate=True)
QCoreApplication.processEvents()
assign_streaks(self.hdf5_path, write=True)
save_atlas(self.hdf5_path, zdim=zdim, left_streak_offset_um=self._atlas_backlash_offset_um())
return True
except Exception as exc:
log_print(f"Automatic atlas reconstruction failed: {exc}", level="warning")
return False
finally:
self._clear_status_message()
[docs]
def update_map_thumbnail(self):
"""Show a saved atlas (entry/atlas/<zdim>) as a thumbnail in the Map preview box.
Renders with the current colormap and fits it within the workflow column so
it never widens that column. Keeps the Atlas group available when the file
has enough data to open the atlas viewer, even before a saved atlas exists.
"""
group = getattr(self, "map_preview_group", None)
if group is None:
return
if not self.hdf5_path:
group.setVisible(False)
return
try:
from coseda.map.streaks import load_atlas
with h5py.File(self.hdf5_path, "r") as f:
names = list(f["entry/atlas"].keys()) if "entry/atlas" in f else []
if not names:
self.map_thumbnail.clear()
self.map_thumbnail.setText("No saved atlas")
group.setVisible(self.open_map_button.isEnabled())
return
zdim = "frame_mean_intensities" if "frame_mean_intensities" in names else names[0]
atlas = load_atlas(self.hdf5_path, zdim=zdim)
except Exception as exc:
log_print(f"Map thumbnail failed: {exc}")
self.map_thumbnail.clear()
self.map_thumbnail.setText("Atlas preview unavailable")
group.setVisible(self.open_map_button.isEnabled())
return
if atlas is None:
self.map_thumbnail.clear()
self.map_thumbnail.setText("No saved atlas")
group.setVisible(self.open_map_button.isEnabled())
return
raster = np.asarray(atlas["raster"], dtype=float)
finite = raster[np.isfinite(raster)]
vmin = float(np.percentile(finite, 1.0)) if finite.size else 0.0
vmax = float(np.percentile(finite, 99.0)) if finite.size else 1.0
if not np.isfinite(vmax) or vmax <= vmin:
vmax = vmin + 1.0
norm = np.clip((raster - vmin) / (vmax - vmin), 0.0, 1.0)
cmap_name = self.cmap_combo.currentText() if hasattr(self, "cmap_combo") else "inferno_r"
try:
cmap = plt.get_cmap(cmap_name)
except Exception:
cmap = plt.get_cmap("inferno_r")
rgba = (cmap(np.nan_to_num(norm)) * 255).astype(np.uint8)
rgba[~np.isfinite(raster)] = (0, 0, 0, 0) # gaps transparent
rgba = np.flipud(rgba) # image top = max y
# Store the full-res image; sizing happens in _fit_atlas_thumbnail once the
# label has its real width (so the physical aspect renders correctly).
self._atlas_base = QPixmap.fromImage(ImageQt(Image.fromarray(rgba, "RGBA")))
self._atlas_extent = atlas["extent"]
group.setVisible(True)
self._fit_atlas_thumbnail() # immediate (fallback width)
QTimer.singleShot(0, self._fit_atlas_thumbnail) # again after layout settles
def _fit_atlas_thumbnail(self):
"""Scale the stored atlas image to the label's width at the correct physical aspect."""
base = getattr(self, "_atlas_base", None)
extent = getattr(self, "_atlas_extent", None)
if base is None or extent is None:
return
w = self.map_thumbnail.width()
if w < 20: # not laid out yet -> estimate from the (visible) workflow column
wg = self.workflow_group.width() if getattr(self, "workflow_group", None) else 0
w = max(80, wg - 16)
x0, x1, y0, y1 = extent
xspan, yspan = abs(x1 - x0), abs(y1 - y0)
aspect = (xspan / yspan) if yspan > 0 else (base.width() / max(base.height(), 1))
h = min(int(w / aspect) if aspect > 0 else base.height(), 400)
w2 = max(1, int(h * aspect))
scaled = base.scaled(w2, h, Qt.AspectRatioMode.IgnoreAspectRatio,
Qt.TransformationMode.SmoothTransformation)
draw_pixmap_scalebar(scaled, xspan) # xspan = physical width (metres)
self.map_thumbnail.setPixmap(scaled)
[docs]
def open_aurora_plot(self):
if not self.hdf5_path:
QMessageBox.warning(self, "No HDF5 file", "Please open an HDF5 file first.")
return
existing = getattr(self, "aurora_plot_window", None)
if existing is not None and existing.isVisible() and existing.hdf5_path != self.hdf5_path:
existing.close()
if existing.isVisible():
existing.raise_()
existing.activateWindow()
return
existing = None
if existing is None or not existing.isVisible():
workspace_files = []
for ini_path in getattr(self, "workspace_ini_paths", {}).values():
try:
_, _, _, _, _, h5file_path = config_to_paths(ini_path)
except Exception:
continue
if h5file_path and os.path.exists(h5file_path):
workspace_files.append({"hdf5_path": h5file_path, "ini_path": ini_path})
if not workspace_files:
workspace_files = [{"hdf5_path": self.hdf5_path, "ini_path": getattr(self, "full_ini_file_path", None)}]
self.aurora_plot_window = AuroraPlotWindow(
self,
hdf5_path=self.hdf5_path,
ini_path=getattr(self, "full_ini_file_path", None),
workspace_files=workspace_files,
)
self.aurora_plot_window.show()
else:
self.aurora_plot_window.raise_()
self.aurora_plot_window.activateWindow()
[docs]
def handle_image_update(self, message):
# Handle any updates from the MapWindow
log_print(f"Received signal from MapWindow: {message}")
# Perform any UI updates or further actions based on the signal
[docs]
def find_strongest_frame(self):
"""Scan nPeaks lazily so the UI stays responsive even for large files."""
self.strongest_frame_index = None
max_peaks = -1
if not self.hdf5_path or not os.path.exists(self.hdf5_path):
self.go_to_strongest_frame_button.setEnabled(False)
return
try:
with h5py.File(self.hdf5_path, 'r') as file:
if 'entry/data/nPeaks' not in file:
self.go_to_strongest_frame_button.setEnabled(False)
return
n_peaks_ds = file['entry/data/nPeaks']
total = min(self.total_frames, n_peaks_ds.shape[0])
chunk_size = 500
for start in range(0, total, chunk_size):
end = min(start + chunk_size, total)
chunk = n_peaks_ds[start:end]
if chunk.size == 0:
continue
local_max = int(chunk.max())
if local_max > max_peaks:
max_peaks = local_max
local_index = int(np.argmax(chunk))
self.strongest_frame_index = start + local_index
QCoreApplication.processEvents()
except Exception as exc:
log_print(f"Error finding strongest frame: {exc}")
self.go_to_strongest_frame_button.setEnabled(self.strongest_frame_index is not None)
[docs]
def go_to_strongest_frame(self):
self.current_frame_index = self.strongest_frame_index
self.update_image()
[docs]
def resizeEvent(self, event):
"""
On window resize, immediately refresh the image to fit the new width
and adjust the container height to maintain the original aspect ratio.
"""
super().resizeEvent(event)
self.update_image()
[docs]
def keyPressEvent(self, event):
"""
Handle left/right arrow keys to navigate frames.
"""
from PyQt6.QtCore import Qt
key = event.key()
if key == Qt.Key.Key_Left:
if self.current_frame_index > 0:
self.previous_frame()
elif key == Qt.Key.Key_Right:
if self.current_frame_index < self.total_frames - 1:
self.next_frame()
else:
super().keyPressEvent(event)