from typing import Optional
from PyQt6.QtCore import QCoreApplication, Qt
from PyQt6.QtWidgets import QApplication, QMessageBox
from coseda.logging_utils import log_print
[docs]
class IndexingMixin:
def _handle_indexing_toggle_request(self, checked: bool, source: Optional[str] = None):
"""Synchronize View-menu action and checkbox, then apply overlay change."""
if self._indexing_toggle_guard:
return
self._indexing_toggle_guard = True
try:
if source != 'action' and hasattr(self, 'show_indexing_results_action'):
self.show_indexing_results_action.blockSignals(True)
self.show_indexing_results_action.setChecked(checked)
self.show_indexing_results_action.blockSignals(False)
if source != 'checkbox' and hasattr(self, 'show_indexing_checkbox'):
self.show_indexing_checkbox.blockSignals(True)
self.show_indexing_checkbox.setChecked(checked)
self.show_indexing_checkbox.blockSignals(False)
finally:
self._indexing_toggle_guard = False
self.on_toggle_indexing_results(checked)
def _set_indexing_controls_checked(self, checked: bool):
"""Update both indexing toggle widgets without triggering overlay logic."""
self._indexing_toggle_guard = True
try:
if hasattr(self, 'show_indexing_results_action'):
self.show_indexing_results_action.setChecked(checked)
if hasattr(self, 'show_indexing_checkbox'):
self.show_indexing_checkbox.setChecked(checked)
finally:
self._indexing_toggle_guard = False
def _update_indexed_navigation_controls(self, enabled: bool):
"""Show or hide indexed-frame navigation buttons based on stream state."""
widget = getattr(self, 'indexed_navigation_widget', None)
if widget is None:
return
widget.setVisible(enabled)
for button in getattr(self, '_indexed_navigation_buttons', []):
button.setEnabled(enabled)
button.setVisible(enabled)
def _preload_indexing_stream(self) -> bool:
"""Parse the selected stream immediately so navigation buttons work on first click."""
if not self.indexing_manager.is_stream_available():
return False
if self.indexing_manager.is_loaded():
return True
return self._ensure_indexing_stream_parsed("Parsing indexing results…")
def _set_status_message(self, message: str, animate: bool = False):
"""Display a status message, optionally animating trailing dots."""
self._status_base_message = message or ""
self._status_anim_phase = 0
label = getattr(self, 'status_label', None)
if label:
label.setText(self._status_base_message)
label.setVisible(bool(message))
if animate and self._status_base_message:
if not self._status_anim_timer.isActive():
self._status_anim_timer.start()
else:
self._status_anim_timer.stop()
def _advance_status_animation(self):
"""Advance the ellipsis animation for the status label."""
if not self._status_base_message:
return
phases = ["", ".", "..", "..."]
self._status_anim_phase = (self._status_anim_phase + 1) % len(phases)
label = getattr(self, 'status_label', None)
if label:
label.setText(f"{self._status_base_message}{phases[self._status_anim_phase]}")
def _clear_status_message(self):
"""Clear any active status message and stop animation."""
self._status_anim_timer.stop()
self._status_base_message = ""
self._status_anim_phase = 0
label = getattr(self, 'status_label', None)
if label:
label.clear()
label.setVisible(False)
def _set_warning_message(self, message: str):
"""Display a persistent warning message in the status box."""
self._status_anim_timer.stop()
self._status_base_message = message or ""
self._status_anim_phase = 0
label = getattr(self, 'status_label', None)
if label:
label.setText(self._status_base_message)
label.setVisible(bool(message))
def _with_busy_status(self, message: str):
"""Context manager to show busy feedback while lengthy tasks run."""
class _BusyIndicator:
def __init__(self, outer, msg):
self.outer = outer
self.msg = msg
def __enter__(self):
self.outer._set_status_message(self.msg)
QCoreApplication.processEvents()
QApplication.setOverrideCursor(Qt.CursorShape.WaitCursor)
def __exit__(self, exc_type, exc, tb):
QApplication.restoreOverrideCursor()
self.outer._clear_status_message()
return _BusyIndicator(self, message)
def _ensure_indexing_overlay_enabled(self):
"""Turn on the indexing overlay if it is currently disabled."""
action = getattr(self, 'show_indexing_results_action', None)
if action is None:
return
if not action.isChecked():
self._handle_indexing_toggle_request(True)
def _ensure_indexing_stream_parsed(self, busy_message: str) -> bool:
"""Ensure the current stream is parsed, showing a busy indicator if it takes time."""
if not self.indexing_manager.has_parser():
QMessageBox.warning(
self,
"CFstreamparser missing",
"CFstreamparser is not available. Install it to view indexing results."
)
return False
if not self.indexing_manager.has_stream_file():
QMessageBox.warning(
self,
"Stream unavailable",
"No CrystFEL stream file is available for the selected run."
)
return False
if self.indexing_manager.is_loaded():
return True
with self._with_busy_status(busy_message):
if self.indexing_manager.ensure_loaded():
return True
error = self.indexing_manager.get_last_error()
detail = f"{error}" if error else "Unknown error."
QMessageBox.critical(
self,
"Stream parse failed",
f"Failed to parse {self.indexing_manager.stream_path}:\n{detail}"
)
return False
def _indexing_debug(self, message: str):
"""Print debug information about the indexing overlay."""
try:
log_print(f"[IndexingOverlay] {message}")
except Exception:
pass
[docs]
def on_indexing_run_stream_changed(self, stream_path: Optional[str]):
"""
Called by the Index & Integrate window whenever the selected run changes.
Updates the available stream path and resets cached overlays.
"""
self.indexing_manager.set_stream_path(stream_path)
enabled = self.indexing_manager.is_stream_available()
self._update_indexed_navigation_controls(enabled)
if hasattr(self, 'show_indexing_results_action'):
self.show_indexing_results_action.setEnabled(enabled)
if not enabled and self.show_indexing_results_action.isChecked():
self._handle_indexing_toggle_request(False)
elif enabled and self.show_indexing_results_action.isChecked():
if self.indexing_manager.ensure_loaded():
self.update_image()
if hasattr(self, 'show_indexing_checkbox'):
self.show_indexing_checkbox.setEnabled(enabled)
if not enabled and self.show_indexing_checkbox.isChecked():
self._handle_indexing_toggle_request(False)
if hasattr(self, "unit_cell_hist_action"):
self.unit_cell_hist_action.setEnabled(enabled)
# Keep the Unit Cell Histogram window synced with the current stream
try:
wnd = getattr(self, "unit_cell_hist_window", None)
if wnd is not None and wnd.isVisible():
wnd.set_stream_path(stream_path, auto_plot=True)
except Exception:
pass
[docs]
def on_toggle_indexing_results(self, checked: bool):
"""Handle enabling/disabling the indexing overlay."""
if checked:
if not self.indexing_manager.has_parser():
QMessageBox.warning(
self,
"CFstreamparser missing",
"CFstreamparser is not available. Install it to view indexing results."
)
self._set_indexing_controls_checked(False)
return
if not self._ensure_indexing_stream_parsed("Parsing indexing results…"):
self._set_indexing_controls_checked(False)
return
self.update_image()
def _ensure_indexing_stream_ready(self) -> bool:
"""Ensure a stream is available and parsed for navigation."""
if not self.indexing_manager.has_parser():
QMessageBox.warning(
self,
"CFstreamparser missing",
"CFstreamparser is not available. Install it to view indexing results."
)
return False
if not self.indexing_manager.has_stream_file():
QMessageBox.warning(self, "No stream", "No CrystFEL stream is available for the current run.")
return False
if not self.indexing_manager.is_loaded():
if not self._ensure_indexing_stream_parsed("Parsing indexing results…"):
return False
if not self.indexing_manager.has_chunks():
QMessageBox.information(self, "No indexed frames", "The selected stream contains no indexed frames yet.")
return False
return True
def _activate_indexed_frame_from_chunk(self, chunk_index: int, chunk, action_label: str):
"""Switch the viewer to the frame corresponding to chunk_index/chunk."""
frame_index = self.indexing_manager.get_frame_index_for_chunk(chunk)
if frame_index is None:
self._indexing_debug(
f"{action_label}: chunk index {chunk_index} could not be mapped to a frame."
)
QMessageBox.warning(
self,
"Frame not found",
"Could not map the indexed pattern to a frame in the loaded HDF5 file."
)
return
if frame_index < 0 or (self.total_frames and frame_index >= self.total_frames):
QMessageBox.warning(
self,
"Frame out of range",
f"Frame {frame_index} is outside the available range."
)
return
self._indexing_debug(
f"{action_label}: chunk {chunk_index} mapped to frame {frame_index}."
)
self.current_frame_index = frame_index
self.frame_slider.setValue(frame_index)
self.frame_entry.setText(str(frame_index + 1))
self.update_image()
def _warn_if_peak_mismatch(self, frame_index: int, h5_peaks, stream_peaks):
"""Warn the user if peaks from HDF5 and stream do not agree."""
valid_h5 = [(fs, ss) for fs, ss in h5_peaks if fs is not None and ss is not None]
valid_stream = [(fs, ss) for fs, ss in stream_peaks if fs is not None and ss is not None]
if not valid_h5 or not valid_stream:
self._last_peak_mismatch_frame = None
return
if len(valid_h5) != len(valid_stream):
if self._last_peak_mismatch_frame == frame_index:
return
self._last_peak_mismatch_frame = frame_index
message = (
"Number of peaks in HDF5 does not match the number of peaks in the CrystFEL stream.\n\n"
f"HDF5 peaks: {len(valid_h5)}\n"
f"Stream peaks: {len(valid_stream)}"
)
QMessageBox.warning(self, "Stream peak mismatch", message)
return
tolerance_px = 0.6
missing, extra = self.indexing_manager.compare_peaks(
valid_h5, valid_stream, tolerance=tolerance_px
)
if missing == 0 and extra == 0:
self._last_peak_mismatch_frame = None
return
if self._last_peak_mismatch_frame == frame_index:
return
self._last_peak_mismatch_frame = frame_index
message = (
"Peaks from the HDF5 data do not match the peaks from the CrystFEL stream.\n\n"
f"HDF5-only peaks: {missing}\n"
f"Stream-only peaks: {extra}\n"
f"Matching tolerance: +/-{tolerance_px:.1f} px"
)
QMessageBox.warning(self, "Stream peak mismatch", message)
[docs]
def show_latest_indexed_frame(self):
"""Jump to the most recently indexed frame based on the current stream."""
if not self._ensure_indexing_stream_ready():
return
chunk_index, chunk = self.indexing_manager.get_last_chunk_with_reflections()
if chunk_index is None or chunk is None:
QMessageBox.warning(
self,
"No reflections",
"No indexed reflections were found in the current stream."
)
return
self._ensure_indexing_overlay_enabled()
self._activate_indexed_frame_from_chunk(chunk_index, chunk, "Latest indexed frame")
[docs]
def show_previous_indexed_frame(self):
"""Jump to the previous indexed frame relative to the current chunk."""
if not self._ensure_indexing_stream_ready():
return
chunk = self.indexing_manager.get_chunk_for_frame(self.current_frame_index)
if chunk is None:
QMessageBox.warning(
self,
"No indexed frame",
"The current frame is not associated with an indexed pattern."
)
return
chunk_index = self.indexing_manager.get_chunk_index(chunk)
if chunk_index is None:
QMessageBox.warning(
self,
"No indexed frame",
"The current frame is not associated with an indexed pattern."
)
return
self._ensure_indexing_overlay_enabled()
target_index, target_chunk = self.indexing_manager.find_chunk_with_reflections_from(chunk_index - 1, -1)
if target_index is None or target_chunk is None:
QMessageBox.information(
self,
"No previous frames",
"There are no earlier indexed frames in this stream."
)
return
self._activate_indexed_frame_from_chunk(target_index, target_chunk, "Previous indexed frame")
[docs]
def show_next_indexed_frame(self):
"""Jump to the next indexed frame relative to the current chunk."""
if not self._ensure_indexing_stream_ready():
return
chunk = self.indexing_manager.get_chunk_for_frame(self.current_frame_index)
if chunk is None:
QMessageBox.warning(
self,
"No indexed frame",
"The current frame is not associated with an indexed pattern."
)
return
chunk_index = self.indexing_manager.get_chunk_index(chunk)
if chunk_index is None:
QMessageBox.warning(
self,
"No indexed frame",
"The current frame is not associated with an indexed pattern."
)
return
self._ensure_indexing_overlay_enabled()
target_index, target_chunk = self.indexing_manager.find_chunk_with_reflections_from(chunk_index + 1, 1)
if target_index is None or target_chunk is None:
QMessageBox.information(
self,
"No later frames",
"There are no later indexed frames in this stream."
)
return
self._activate_indexed_frame_from_chunk(target_index, target_chunk, "Next indexed frame")