From 1d7fb19008d48d1ec10ea28367a3da77a4faebdc Mon Sep 17 00:00:00 2001 From: fdrgsp Date: Fri, 1 May 2026 19:18:25 -0400 Subject: [PATCH 1/5] feat: wip --- src/cali/gui/_cali_gui.py | 50 +- src/cali/gui/_runs_panel.py | 984 ++++++++++++++++++++++-------------- 2 files changed, 660 insertions(+), 374 deletions(-) diff --git a/src/cali/gui/_cali_gui.py b/src/cali/gui/_cali_gui.py index 23d980ae..61f891f5 100644 --- a/src/cali/gui/_cali_gui.py +++ b/src/cali/gui/_cali_gui.py @@ -347,6 +347,9 @@ def __init__( self._fov_table.doubleClicked.connect(self._on_fov_double_click) self._runs_panel.runSelected.connect(self._on_run_item_selected) + self._runs_panel.segmentationSelected.connect( + self._on_saved_segmentation_selected + ) self._runs_panel.settingsDeleted.connect(self._on_settings_deleted) # connect the roiSelected signal from the graphs to the image viewer so we can @@ -994,7 +997,7 @@ def _update_gui_settings( self._runs_panel._runs_list.setCurrentRow(0) # emit runSelected signal for the first run if (first_item := self._runs_panel._runs_list.item(0)) is not None: - self._runs_panel._on_item_clicked(first_item) + self._runs_panel._on_run_item_clicked(first_item) # load plate plan data if experiment is None: experiment = Experiment.load_from_database(database_path, load_data=False) @@ -2420,6 +2423,51 @@ def _on_run_item_selected(self, run_id: int) -> None: show_error_dialog(self, f"Failed to load run settings: {e}") cali_logger.error(f"❌ Failed to load run #{run_id}: {e}") + def _on_saved_segmentation_selected(self, detection_settings_id: int) -> None: + """Handle selection of an orphan (saved) segmentation in the runs panel. + + Loads the detection parameters into the detection widget and refreshes + the image viewer so the labels reflect this segmentation. + """ + if self._database_path is None: + return + + try: + d_settings = DetectionSettings.load_from_database( + self._database_path, id=detection_settings_id + ) + assert isinstance(d_settings, DetectionSettings) + if d_settings.method == "cellpose": + self._detection_wdg.setValue( + CellposeSettingsData( + model_type=d_settings.model_type, + model_path=d_settings.custom_model, + diameter=d_settings.diameter, + cellprob_threshold=d_settings.cellprob_threshold, + flow_threshold=d_settings.flow_threshold, + min_size=d_settings.min_size, + normalize=d_settings.normalize, + batch_size=d_settings.batch_size, + use_gpu=d_settings.use_gpu, + ) + ) + elif d_settings.method == "imported_labels": # pragma: no cover + self._detection_wdg.setValue(method="imported_labels") + self._detection_wdg._imported_labels_wdg.set_detection_settings_id( + d_settings.id + ) + + cali_logger.info(f"✅ Selected saved segmentation #{detection_settings_id}") + + # Refresh the image viewer to show the segmentation labels + self._on_fov_table_selection_changed() + + except Exception as e: + show_error_dialog(self, f"Failed to load saved segmentation: {e}") + cali_logger.error( + f"❌ Failed to load saved segmentation #{detection_settings_id}: {e}" + ) + def _set_splitter_sizes(self) -> None: """Set the initial sizes for the splitters.""" splitter_and_sizes = ( diff --git a/src/cali/gui/_runs_panel.py b/src/cali/gui/_runs_panel.py index 3d09f38d..40e84eb6 100644 --- a/src/cali/gui/_runs_panel.py +++ b/src/cali/gui/_runs_panel.py @@ -3,20 +3,26 @@ from __future__ import annotations from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any, NamedTuple from qtpy.QtCore import QEvent, QObject, Qt, Signal from qtpy.QtWidgets import ( + QCheckBox, + QDialog, + QDialogButtonBox, QGroupBox, QHBoxLayout, + QLabel, QListWidget, QListWidgetItem, QMessageBox, QPushButton, QSizePolicy, + QSplitter, QVBoxLayout, QWidget, ) +from sqlalchemy import func from sqlmodel import select from superqt import QIconifyIcon from superqt.utils import signals_blocked @@ -29,20 +35,85 @@ from sqlmodel import Session +class _DetectionSummary(NamedTuple): + """Summary info for a DetectionSettings row, used in dialogs and the saved list.""" + + detection_id: int + method: str + model_type: str | None + run_count: int + roi_count: int + fov_count: int + + def label(self) -> str: + """Render a single-line description for UI.""" + head = f"Detection #{self.detection_id} — {self.method}" + if self.model_type: + head += f" / {self.model_type}" + run_word = "run" if self.run_count == 1 else "runs" + roi_word = "ROI" if self.roi_count == 1 else "ROIs" + fov_word = "FOV" if self.fov_count == 1 else "FOVs" + return ( + f"{head} ({self.run_count} {run_word}, " + f"{self.roi_count} {roi_word} across {self.fov_count} {fov_word})" + ) + + +class _DetectionKeepDialog(QDialog): + """Dialog asking the user which detections to keep when deleting all runs.""" + + def __init__( + self, summaries: list[_DetectionSummary], parent: QWidget | None = None + ) -> None: + super().__init__(parent) + self.setWindowTitle("Delete All Runs") + layout = QVBoxLayout(self) + layout.addWidget( + QLabel( + "All runs will be deleted.\n" + "Tick any segmentations you want to keep — unticked ones will be " + "deleted along with their ROIs." + ) + ) + self._checkboxes: dict[int, QCheckBox] = {} + for summary in summaries: + cb = QCheckBox(summary.label()) + self._checkboxes[summary.detection_id] = cb + layout.addWidget(cb) + if not summaries: + layout.addWidget(QLabel("(no detection settings to keep)")) + + button_box = QDialogButtonBox( + QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel + ) + button_box.accepted.connect(self.accept) + button_box.rejected.connect(self.reject) + layout.addWidget(button_box) + + def kept_detection_ids(self) -> set[int]: + return {did for did, cb in self._checkboxes.items() if cb.isChecked()} + + class _RunsPanel(QGroupBox): """Panel that displays analysis and detection runs. - This widget displays a list of all analysis runs stored in the database. + This widget displays a list of all analysis runs stored in the database, + plus a section listing orphan ("saved") segmentations — DetectionSettings + rows kept around without an associated CaliResult. Signals ------- runSelected : int Emitted when a run is selected, passes the CaliResult ID - settingsChanged : None - Emitted when detection settings may have changed (e.g., after deletion) + segmentationSelected : int + Emitted when a saved (orphan) segmentation is selected, passes the + DetectionSettings ID + settingsDeleted : None + Emitted when settings may have changed (e.g. after deletion) """ runSelected = Signal(int) + segmentationSelected = Signal(int) settingsDeleted = Signal() def __init__(self, parent: QWidget | None = None) -> None: @@ -56,28 +127,51 @@ def __init__(self, parent: QWidget | None = None) -> None: layout.setContentsMargins(10, 10, 10, 10) layout.setSpacing(5) - # List widget for runs + # Splitter so the user can resize runs vs saved segmentations + self._splitter = QSplitter(Qt.Orientation.Vertical) + + # Runs list self._runs_list = QListWidget() self._runs_list.setAlternatingRowColors(True) self._runs_list.setToolTip( "Click on a run to load its analysis and detection settings" ) + self._splitter.addWidget(self._runs_list) + + # Saved segmentations section + saved_container = QWidget() + saved_layout = QVBoxLayout(saved_container) + saved_layout.setContentsMargins(0, 0, 0, 0) + saved_layout.setSpacing(2) + saved_layout.addWidget(QLabel("Saved Segmentations")) + self._saved_segs_list = QListWidget() + self._saved_segs_list.setAlternatingRowColors(True) + self._saved_segs_list.setToolTip( + "Segmentations kept after their runs were deleted. " + "Click to view labels or reuse in a new run." + ) + saved_layout.addWidget(self._saved_segs_list) + self._splitter.addWidget(saved_container) + self._splitter.setStretchFactor(0, 4) + self._splitter.setStretchFactor(1, 1) - layout.addWidget(self._runs_list) + layout.addWidget(self._splitter) # Buttons layout buttons_layout = QHBoxLayout() buttons_layout.addStretch() # Push buttons to the right - # Delete selected run button + # Delete selected button (works for either list) self._delete_btn = QPushButton("Delete Selected") self._delete_btn.setIcon(QIconifyIcon("mdi:delete", color=RED)) - self._delete_btn.setToolTip("Delete the selected run from the database") - self._delete_btn.clicked.connect(self._delete_selected_run) - self._delete_btn.setEnabled(False) # Disabled by default + self._delete_btn.setToolTip( + "Delete the selected run or saved segmentation from the database" + ) + self._delete_btn.clicked.connect(self._delete_selected) + self._delete_btn.setEnabled(False) buttons_layout.addWidget(self._delete_btn) - # Clear all runs button + # Clear all button self._clear_all_btn = QPushButton("Delete All") self._clear_all_btn.setIcon(QIconifyIcon("mdi:delete-forever", color=RED)) self._clear_all_btn.setToolTip("Delete all runs from the database") @@ -86,125 +180,99 @@ def __init__(self, parent: QWidget | None = None) -> None: layout.addLayout(buttons_layout) - # Connect selection change to enable/disable delete button - self._runs_list.itemSelectionChanged.connect(self._on_selection_changed) - self._runs_list.itemClicked.connect(self._on_item_clicked) + # Selection wiring — clicking one list deselects the other + self._runs_list.itemSelectionChanged.connect(self._on_runs_selection_changed) + self._runs_list.itemClicked.connect(self._on_run_item_clicked) + self._saved_segs_list.itemSelectionChanged.connect( + self._on_saved_segs_selection_changed + ) + self._saved_segs_list.itemClicked.connect(self._on_saved_seg_clicked) - # Allow deselecting by clicking empty area in list + # Allow deselecting by clicking empty area in either list self._runs_list.viewport().installEventFilter(self) + self._saved_segs_list.viewport().installEventFilter(self) # Set size policy self.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Expanding) + # ------------------------------------------------------------------ public API + def clear(self) -> None: - """Clear the runs list.""" + """Clear both lists.""" self._runs_list.clear() + self._saved_segs_list.clear() def database_path(self) -> Path | None: - """Get the current database path. - - Returns - ------- - Path | None - Path to the database file or None if not set - """ + """Get the current database path.""" return self._database_path def set_database_path(self, db_path: Path | str) -> None: - """Set the database path and reload runs. - - Parameters - ---------- - db_path : Path | None - Path to the database file - """ + """Set the database path and reload runs.""" if isinstance(db_path, str): db_path = Path(db_path) - self._database_path = db_path self.refresh_runs() def refresh_runs(self) -> None: - """Refresh the list of runs from the database.""" + """Refresh the runs list and the saved-segmentations list.""" self.clear() if self._database_path is None or not self._database_path.exists(): return try: - from sqlmodel import Session, create_engine, select + from sqlmodel import Session, create_engine engine = create_engine(f"sqlite:///{self._database_path}") with Session(engine) as session: - # Join CaliResult with DetectionSettings to avoid N+1 queries - # Order by created_at ascending (most recent last) + # Runs stmt = ( select(CaliResult, DetectionSettings) .where(CaliResult.detection_settings_id == DetectionSettings.id) .order_by(CaliResult.created_at) ) - results = session.exec(stmt).all() - - for result, detection_settings in results: + for result, detection_settings in session.exec(stmt).all(): self._add_run_item(result, detection_settings) + # Saved (orphan) segmentations: DetectionSettings without any CaliResult + used_ids_stmt = select(CaliResult.detection_settings_id).where( + CaliResult.detection_settings_id.is_not(None) # type: ignore[union-attr] + ) + orphan_stmt = ( + select(DetectionSettings) + .where(DetectionSettings.id.not_in(used_ids_stmt)) # type: ignore[union-attr] + .order_by(DetectionSettings.id) + ) + for d_settings in session.exec(orphan_stmt).all(): + summary = self._build_summary(session, d_settings) + self._add_saved_seg_item(summary) + engine.dispose(close=True) except Exception as e: cali_logger.error(f"Error loading runs: {e}") def select_run_by_index(self, idx: int, block_signals: bool = False) -> None: - """Select a run by its index in the list. - - Parameters - ---------- - idx : int - Index of the run to select - block_signals : bool - Whether to block signals during selection (default: False) - """ + """Select a run by its index in the list.""" if 0 <= idx < self._runs_list.count(): item = self._runs_list.item(idx) if item: - if block_signals: - self._runs_list.setCurrentItem(item) - else: - self._runs_list.setCurrentItem(item) - self._on_item_clicked(item) + self._runs_list.setCurrentItem(item) + if not block_signals: + self._on_run_item_clicked(item) def select_run_by_id(self, run_id: int, block_signals: bool = False) -> None: - """Select a run by its CaliResult ID. - - Parameters - ---------- - run_id : int - CaliResult ID of the run to select - block_signals : bool - Whether to block signals during selection (default: False) - """ + """Select a run by its CaliResult ID.""" for i in range(self._runs_list.count()): item = self._runs_list.item(i) if item and item.data(Qt.ItemDataRole.UserRole) == run_id: - if block_signals: - self._runs_list.setCurrentItem(item) - else: - self._runs_list.setCurrentItem(item) - self._on_item_clicked(item) + self._runs_list.setCurrentItem(item) + if not block_signals: + self._on_run_item_clicked(item) return def get_run_id_by_index(self, idx: int) -> int | None: - """Get the CaliResult ID of the run at the given index. - - Parameters - ---------- - idx : int - Index of the run in the list - - Returns - ------- - int | None - CaliResult ID of the run, or None if index is invalid - """ + """Get the CaliResult ID of the run at the given index.""" if 0 <= idx < self._runs_list.count(): item = self._runs_list.item(idx) if item: @@ -212,135 +280,71 @@ def get_run_id_by_index(self, idx: int) -> int | None: return None def get_selected_run_id(self) -> int | None: - """Get the ID of the currently selected run. - - Returns - ------- - int | None - CaliResult ID of the selected run, or None if no run selected - """ + """Get the ID of the currently selected run, or None if no run selected.""" current_item = self._runs_list.currentItem() - if current_item is None: + if current_item is None or not current_item.isSelected(): return None - return current_item.data(Qt.ItemDataRole.UserRole) # type: ignore def get_selected_detection_settings_id(self) -> int | None: - """Get the detection settings ID from the currently selected run. + """Get the detection settings ID currently in focus. - Returns - ------- - int | None - Detection settings ID of the selected run, or None if no run selected + Returns the detection_settings_id from the selected run, or — if no run + is selected — the detection_settings_id of the selected saved + segmentation. Returns None if neither list has a selection. """ - current_item = self._runs_list.currentItem() - if current_item is None or self._database_path is None: - return None + # Saved seg has priority only when no run is selected + run_item = self._runs_list.currentItem() + if run_item is not None and run_item.isSelected(): + run_id = run_item.data(Qt.ItemDataRole.UserRole) + if run_id is None or self._database_path is None: + return None + try: + from sqlmodel import Session, create_engine - run_id = current_item.data(Qt.ItemDataRole.UserRole) - if self._database_path is None: - return None + engine = create_engine( + f"sqlite:///{self._database_path}", + connect_args={"timeout": 30.0, "check_same_thread": False}, + pool_pre_ping=True, + ) + with Session(engine) as session: + result = session.get(CaliResult, run_id) + detection_id = result.detection_settings_id if result else None + engine.dispose(close=True) + return detection_id + except Exception as e: + cali_logger.error(f"Failed to get detection settings ID: {e}") + return None - try: - from sqlmodel import Session, create_engine + seg_item = self._saved_segs_list.currentItem() + if seg_item is not None and seg_item.isSelected(): + return seg_item.data(Qt.ItemDataRole.UserRole) # type: ignore - engine = create_engine( - f"sqlite:///{self._database_path}", - connect_args={"timeout": 30.0, "check_same_thread": False}, - pool_pre_ping=True, - ) - with Session(engine) as session: - result = session.get(CaliResult, run_id) - detection_id = result.detection_settings_id if result else None - engine.dispose(close=True) - return detection_id - except Exception as e: - cali_logger.error(f"Failed to get detection settings ID: {e}") + return None + + def get_selected_saved_segmentation_id(self) -> int | None: + """Get the DetectionSettings ID of the selected saved segmentation, if any.""" + item = self._saved_segs_list.currentItem() + if item is None or not item.isSelected(): return None + return item.data(Qt.ItemDataRole.UserRole) # type: ignore def get_detection_settings_ids(self) -> list[int]: - """Get all unique detection settings IDs from database. - - Queries the DetectionSettings table directly to find all available - detection settings, regardless of whether they're in a CaliResult. - - Returns - ------- - list[int] - Sorted list of unique detection settings IDs - """ - if self._database_path is None: - return [] - - try: - from sqlmodel import Session, create_engine, select - - from cali.sqlmodel._model import DetectionSettings - - engine = create_engine( - f"sqlite:///{self._database_path}", - connect_args={"timeout": 30.0, "check_same_thread": False}, - pool_pre_ping=True, - ) - with Session(engine) as session: - # Get all detection settings IDs directly from the table - stmt = select(DetectionSettings.id) - results = session.exec(stmt).all() - ids = {r for r in results if r is not None} - engine.dispose(close=True) - return sorted(ids) - except Exception as e: - cali_logger.error(f"Failed to get detection settings IDs: {e}") - return [] + """Get all unique detection settings IDs from the database.""" + return self._fetch_ids(DetectionSettings.id) def get_extraction_settings_ids(self) -> list[int]: - """Get all unique extraction settings IDs from database. + """Get all unique extraction settings IDs from the database.""" + from cali.sqlmodel._model import ExtractionSettings - Queries the ExtractionSettings table directly to find all available - extraction settings, regardless of whether they're in a CaliResult. - - Returns - ------- - list[int] - Sorted list of unique extraction settings IDs - """ - if self._database_path is None: - return [] - - try: - from sqlmodel import Session, create_engine, select - - from cali.sqlmodel._model import ExtractionSettings - - engine = create_engine( - f"sqlite:///{self._database_path}", - connect_args={"timeout": 30.0, "check_same_thread": False}, - pool_pre_ping=True, - ) - with Session(engine) as session: - # Get all extraction settings IDs directly from the table - stmt = select(ExtractionSettings.id) - results = session.exec(stmt).all() - ids = {r for r in results if r is not None} - engine.dispose(close=True) - return sorted(ids) - except Exception as e: - cali_logger.error(f"Failed to get extraction settings IDs: {e}") - return [] + return self._fetch_ids(ExtractionSettings.id) def get_analysis_settings_ids(self) -> list[int]: - """Get all unique analysis settings IDs from runs. - - Returns - ------- - list[int] - Sorted list of unique analysis settings IDs - """ + """Get all unique analysis settings IDs from runs.""" if self._database_path is None: return [] - try: - from sqlmodel import Session, create_engine, select + from sqlmodel import Session, create_engine engine = create_engine( f"sqlite:///{self._database_path}", @@ -348,10 +352,8 @@ def get_analysis_settings_ids(self) -> list[int]: pool_pre_ping=True, ) with Session(engine) as session: - # Get all unique analysis settings IDs stmt = select(CaliResult.analysis_settings_id).distinct() - results = session.exec(stmt).all() - ids = {r for r in results if r is not None} + ids = {r for r in session.exec(stmt).all() if r is not None} engine.dispose(close=True) return sorted(ids) except Exception as e: @@ -359,18 +361,11 @@ def get_analysis_settings_ids(self) -> list[int]: return [] def get_run_ids(self) -> list[int]: - """Get all run IDs from database. - - Returns - ------- - list[int] - Sorted list of all run IDs - """ + """Get all run IDs from the database.""" if self._database_path is None: return [] - try: - from sqlmodel import Session, create_engine, select + from sqlmodel import Session, create_engine engine = create_engine( f"sqlite:///{self._database_path}", @@ -379,10 +374,8 @@ def get_run_ids(self) -> list[int]: ) try: with Session(engine) as session: - # Get all run IDs stmt = select(CaliResult.id) - results = session.exec(stmt).all() - ids = [r for r in results if r is not None] + ids = [r for r in session.exec(stmt).all() if r is not None] return sorted(ids) finally: engine.dispose(close=True) @@ -399,22 +392,13 @@ def highlight_run_by_settings( """Highlight the run that matches detection, extraction, and analysis settings. If no exact match is found, deselect all runs. - - Parameters - ---------- - detection_id : int | None - Detection settings ID to match - extraction_id : int | None - Extraction settings ID to match (None for detection-only runs) - analysis_id : int | None - Analysis settings ID to match (None for runs without analysis) """ if self._database_path is None: return try: from sqlalchemy import desc - from sqlmodel import Session, create_engine, select + from sqlmodel import Session, create_engine engine = create_engine( f"sqlite:///{self._database_path}", @@ -422,7 +406,6 @@ def highlight_run_by_settings( pool_pre_ping=True, ) with Session(engine) as session: - # Find run with matching settings query = select(CaliResult) if detection_id is not None: query = query.where( @@ -435,13 +418,11 @@ def highlight_run_by_settings( if analysis_id is not None: query = query.where(CaliResult.analysis_settings_id == analysis_id) - # Order by created_at desc and take first query = query.order_by(desc(CaliResult.created_at)) matching_run = session.exec(query).first() engine.dispose(close=True) - # Find and select the matching item in the list if matching_run: for i in range(self._runs_list.count()): item = self._runs_list.item(i) @@ -450,27 +431,119 @@ def highlight_run_by_settings( self._runs_list.setCurrentItem(item) return - # No match found - deselect all self._runs_list.clearSelection() except Exception as e: cali_logger.error(f"Failed to highlight run by settings: {e}") + # ------------------------------------------------------------- internal helpers + + def _fetch_ids(self, column: Any) -> list[int]: + """Fetch a sorted, unique list of non-null IDs from a single column.""" + if self._database_path is None: + return [] + try: + from sqlmodel import Session, create_engine + + engine = create_engine( + f"sqlite:///{self._database_path}", + connect_args={"timeout": 30.0, "check_same_thread": False}, + pool_pre_ping=True, + ) + with Session(engine) as session: + ids = {r for r in session.exec(select(column)).all() if r is not None} + engine.dispose(close=True) + return sorted(ids) + except Exception as e: + cali_logger.error(f"Failed to fetch IDs: {e}") + return [] + + def _build_summary( + self, session: Session, d_settings: DetectionSettings + ) -> _DetectionSummary: + """Compute counts (runs / ROIs / FOVs) for a DetectionSettings row.""" + from cali.sqlmodel._model import ROI + + did = d_settings.id + assert did is not None + + run_count = session.exec( + select(func.count()) + .select_from(CaliResult) + .where(CaliResult.detection_settings_id == did) + ).one() + roi_count = session.exec( + select(func.count()) + .select_from(ROI) + .where(ROI.detection_settings_id == did) + ).one() + fov_count = session.exec( + select(func.count(func.distinct(ROI.fov_id))).where( + ROI.detection_settings_id == did + ) + ).one() + + return _DetectionSummary( + detection_id=did, + method=d_settings.method, + model_type=d_settings.model_type, + run_count=int(run_count or 0), + roi_count=int(roi_count or 0), + fov_count=int(fov_count or 0), + ) + + def _all_detection_summaries(self) -> list[_DetectionSummary]: + """Build summaries for every DetectionSettings row in the DB.""" + if self._database_path is None: + return [] + try: + from sqlmodel import Session, create_engine + + engine = create_engine( + f"sqlite:///{self._database_path}", + connect_args={"timeout": 30.0, "check_same_thread": False}, + pool_pre_ping=True, + ) + with Session(engine) as session: + rows = session.exec( + select(DetectionSettings).order_by(DetectionSettings.id) + ).all() + summaries = [self._build_summary(session, d) for d in rows] + engine.dispose(close=True) + return summaries + except Exception as e: + cali_logger.error(f"Failed to compute detection summaries: {e}") + return [] + + def _count_runs_using_detection(self, detection_id: int) -> int: + """Count CaliResults referencing the given detection_settings_id.""" + if self._database_path is None: + return 0 + try: + from sqlmodel import Session, create_engine + + engine = create_engine( + f"sqlite:///{self._database_path}", + connect_args={"timeout": 30.0, "check_same_thread": False}, + pool_pre_ping=True, + ) + with Session(engine) as session: + count = session.exec( + select(func.count()) + .select_from(CaliResult) + .where(CaliResult.detection_settings_id == detection_id) + ).one() + engine.dispose(close=True) + return int(count or 0) + except Exception as e: + cali_logger.error(f"Failed to count runs using detection: {e}") + return 0 + def _add_run_item( self, result: CaliResult, detection_settings: DetectionSettings ) -> None: - """Add a run item to the list. - - Parameters - ---------- - result : CaliResult - The analysis result to add - detection_settings : DetectionSettings - The detection settings associated with the result - """ - # Format the display text + """Add a run item to the list.""" created_at = result.created_at.strftime("%Y-%m-%d %H:%M:%S") - # Include milliseconds (trim to 3 decimal places) last_modified = result.last_modified.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] d_id = result.detection_settings_id @@ -484,7 +557,6 @@ def _add_run_item( extraction_icon = "❌" if result.extraction_settings_id is None else "✅" extraction_incomplete = "" if result.extraction_settings_id is not None: - # Check if extraction is incomplete (detected != extracted) detected = set(result.positions_detected or []) extracted = set(result.positions_extracted or []) if detected and extracted and len(detected) != len(extracted): @@ -499,16 +571,10 @@ def _add_run_item( analysis_icon = "❌" if result.analysis_settings_id is None else "✅" analysis_incomplete = "" if result.analysis_settings_id is not None: - # Check if analysis is incomplete (detected != analyzed) detected = set(result.positions_detected or []) analyzed = set(result.positions_analyzed or []) if detected and analyzed and len(detected) != len(analyzed): analysis_incomplete = " ⚠️" - # Check if analysis is incomplete (extracted != analyzed) - # extracted = set(result.positions_extracted or []) - # analyzed = set(result.positions_analyzed or []) - # if extracted and analyzed and len(extracted) != len(analyzed): - # analysis_incomplete = " ⚠️" item_text += ( f" {analysis_icon} Analysis ID: {result.analysis_settings_id}" @@ -516,7 +582,6 @@ def _add_run_item( ) item = QListWidgetItem(item_text) - # item.setIcon(icon(MDI6.run_fast)) item.setData(Qt.ItemDataRole.UserRole, result.id) item.setToolTip( @@ -533,13 +598,62 @@ def _add_run_item( self._runs_list.addItem(item) - def _on_selection_changed(self) -> None: - """Handle selection change to enable/disable delete button.""" - has_selection = len(self._runs_list.selectedItems()) > 0 + def _add_saved_seg_item(self, summary: _DetectionSummary) -> None: + """Add an orphan-detection item to the saved segmentations list.""" + item = QListWidgetItem(summary.label()) + item.setData(Qt.ItemDataRole.UserRole, summary.detection_id) + item.setToolTip( + f"Saved segmentation\n" + f"Detection Settings ID: {summary.detection_id}\n" + f"Method: {summary.method}\n" + f"Model: {summary.model_type}\n" + f"ROIs: {summary.roi_count} across {summary.fov_count} FOV(s)" + ) + self._saved_segs_list.addItem(item) + + # ------------------------------------------------------------------ selection + + def _on_runs_selection_changed(self) -> None: + has_runs = len(self._runs_list.selectedItems()) > 0 + if has_runs: + with signals_blocked(self._saved_segs_list): + self._saved_segs_list.clearSelection() + self._update_delete_button() + + def _on_saved_segs_selection_changed(self) -> None: + has_segs = len(self._saved_segs_list.selectedItems()) > 0 + if has_segs: + with signals_blocked(self._runs_list): + self._runs_list.clearSelection() + self._update_delete_button() + + def _update_delete_button(self) -> None: + has_selection = bool(self._runs_list.selectedItems()) or bool( + self._saved_segs_list.selectedItems() + ) self._delete_btn.setEnabled(has_selection) + def _on_run_item_clicked(self, item: QListWidgetItem) -> None: + run_id = item.data(Qt.ItemDataRole.UserRole) + if run_id is not None: + self.runSelected.emit(run_id) + + def _on_saved_seg_clicked(self, item: QListWidgetItem) -> None: + detection_id = item.data(Qt.ItemDataRole.UserRole) + if detection_id is not None: + self.segmentationSelected.emit(detection_id) + + # -------------------------------------------------------------------- delete + + def _delete_selected(self) -> None: + """Delete whichever item (run or saved segmentation) is currently selected.""" + if self._saved_segs_list.selectedItems(): + self._delete_selected_saved_segmentation() + else: + self._delete_selected_run() + def _delete_selected_run(self) -> None: - """Delete the selected run from the database.""" + """Delete the selected run, asking about segmentation when relevant.""" current_item = self._runs_list.currentItem() if current_item is None: return @@ -548,55 +662,151 @@ def _delete_selected_run(self) -> None: if run_id is None: return - # Confirm deletion - reply = QMessageBox.warning( - self, - "Confirm Deletion", - f"Are you sure you want to delete Run #{run_id}?\n\n" - "This action cannot be undone.", - QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, - QMessageBox.StandardButton.No, - ) + # If this run is the only one using its detection, give the user a choice + detection_id = self._get_detection_id_for_run(run_id) + keep_detection = False + if ( + detection_id is not None + and self._count_runs_using_detection(detection_id) == 1 + ): + choice = self._ask_keep_or_delete_segmentation(run_id, detection_id) + if choice is None: + return # cancelled + keep_detection = choice + else: + reply = QMessageBox.warning( + self, + "Confirm Deletion", + f"Are you sure you want to delete Run #{run_id}?\n\n" + "This action cannot be undone.", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.No, + ) + if reply != QMessageBox.StandardButton.Yes: + return - if reply == QMessageBox.StandardButton.Yes: - self._delete_run_from_database(run_id) - self.refresh_runs() - self.settingsDeleted.emit() # Notify that settings may have changed + self._delete_run_from_database(run_id, keep_detection=keep_detection) + self.refresh_runs() + self.settingsDeleted.emit() + if keep_detection: + cali_logger.info( + f"🚮 Deleted Run #{run_id}; kept segmentation #{detection_id}." + ) + else: cali_logger.info(f"🚮 Deleted Run #{run_id} from database.") - def _clear_all_runs(self) -> None: - """Delete all runs from the database.""" - if self._runs_list.count() == 0: + def _delete_selected_saved_segmentation(self) -> None: + """Delete an orphan segmentation (DetectionSettings + its ROIs/Masks).""" + item = self._saved_segs_list.currentItem() + if item is None: + return + detection_id = item.data(Qt.ItemDataRole.UserRole) + if detection_id is None: return - # Confirm clearing all reply = QMessageBox.warning( self, - "Confirm Clear All", - "Are you sure you want to delete ALL runs from the database?\n\n" - "This will permanently delete all analysis results and detection " - "settings.\nThis action cannot be undone.", + "Delete Saved Segmentation", + f"Delete saved segmentation #{detection_id}?\n\n" + "All ROIs and masks created by this detection will be permanently " + "removed.\nThis action cannot be undone.", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No, ) + if reply != QMessageBox.StandardButton.Yes: + return + + self._delete_detection_data(detection_id) + self.refresh_runs() + self.settingsDeleted.emit() + cali_logger.info(f"🚮 Deleted saved segmentation #{detection_id}.") + + def _ask_keep_or_delete_segmentation( + self, run_id: int, detection_id: int + ) -> bool | None: + """Ask the user whether to keep the segmentation or delete it too. + + Returns + ------- + bool | None + True to keep, False to delete, None if the user cancelled. + """ + msg = QMessageBox(self) + msg.setIcon(QMessageBox.Icon.Warning) + msg.setWindowTitle("Confirm Deletion") + msg.setText( + f"Run #{run_id} is the only run using Detection #{detection_id}.\n\n" + "Keep the segmentation (ROIs + masks) for future use, or delete " + "everything?" + ) + keep_btn = msg.addButton("Keep segmentation", QMessageBox.ButtonRole.AcceptRole) + delete_btn = msg.addButton( + "Delete everything", QMessageBox.ButtonRole.DestructiveRole + ) + cancel_btn = msg.addButton("Cancel", QMessageBox.ButtonRole.RejectRole) + msg.setDefaultButton(cancel_btn) + msg.exec() + clicked = msg.clickedButton() + if clicked is cancel_btn: + return None + if clicked is keep_btn: + return True + if clicked is delete_btn: + return False + return None + + def _clear_all_runs(self) -> None: + """Delete all runs, with a per-detection keep/delete dialog.""" + if self._runs_list.count() == 0 and self._saved_segs_list.count() == 0: + return + + summaries = self._all_detection_summaries() + dialog = _DetectionKeepDialog(summaries, self) + if dialog.exec() != QDialog.DialogCode.Accepted: + return - if reply == QMessageBox.StandardButton.Yes: - self._clear_all_from_database() - self.refresh_runs() - self.settingsDeleted.emit() # Notify that settings may have changed + kept_ids = dialog.kept_detection_ids() + self._clear_all_from_database(keep_detection_ids=kept_ids) + self.refresh_runs() + self.settingsDeleted.emit() + if kept_ids: + cali_logger.info( + f"🚮 Deleted all runs; kept segmentations: {sorted(kept_ids)}." + ) + else: cali_logger.info("🚮 Deleted ALL runs from database.") - def _delete_run_from_database(self, run_id: int) -> None: - """Delete a specific run from the database with smart cascading. + # -------------------------------------------------------- DB delete operations - Deletes the run and cleans up orphaned settings and ROIs: - - Deletes DetectionSettings if no other run uses them (and their ROIs) - - Deletes AnalysisSettings if no other run uses them + def _get_detection_id_for_run(self, run_id: int) -> int | None: + if self._database_path is None: + return None + try: + from sqlmodel import Session, create_engine + + engine = create_engine(f"sqlite:///{self._database_path}") + with Session(engine) as session: + result = session.get(CaliResult, run_id) + detection_id = result.detection_settings_id if result else None + engine.dispose(close=True) + return detection_id + except Exception as e: + cali_logger.error(f"Failed to look up detection for run {run_id}: {e}") + return None + + def _delete_run_from_database( + self, run_id: int, *, keep_detection: bool = False + ) -> None: + """Delete a run, optionally preserving its detection segmentation. Parameters ---------- run_id : int - The ID of the CaliResult to delete + The CaliResult ID to delete. + keep_detection : bool + If True, leave the DetectionSettings row + ROIs + masks in place + even if no other run references them (they become an orphan + "saved segmentation"). """ if self._database_path is None: return @@ -606,7 +816,6 @@ def _delete_run_from_database(self, run_id: int) -> None: engine = create_engine(f"sqlite:///{self._database_path}") with Session(engine) as session: - # Get the result before deleting to capture its settings IDs result = session.get(CaliResult, run_id) if not result: return @@ -614,27 +823,40 @@ def _delete_run_from_database(self, run_id: int) -> None: detection_id = result.detection_settings_id analysis_id = result.analysis_settings_id - # Delete the analysis result (cascades to Traces via relationship) + # Delete the analysis result (cascades to Traces via FK) session.delete(result) session.commit() - # Clean up orphaned settings and ROIs - self._cleanup_orphaned_data(session, detection_id, analysis_id) + # Clean up orphaned settings (and ROIs unless keep_detection) + self._cleanup_orphaned_data( + session, + detection_id, + analysis_id, + keep_detection=keep_detection, + ) engine.dispose(close=True) except Exception as e: QMessageBox.warning(self, "Error", f"Failed to delete run: {e}") - def _clear_all_from_database(self) -> None: - """Delete all runs from the database with complete cleanup. + def _clear_all_from_database( + self, keep_detection_ids: set[int] | None = None + ) -> None: + """Delete all runs; optionally preserve specific detection segmentations. - Deletes all CaliResults, ROIs, and ALL settings (including orphaned ones). - This ensures a completely clean database state. + Parameters + ---------- + keep_detection_ids : set[int] | None + DetectionSettings IDs to preserve (along with their ROIs/Masks). + All other detections — and their ROIs — will be deleted. + Extraction and analysis settings are always deleted (run-scoped). """ if self._database_path is None: return + keep = set(keep_detection_ids or set()) + try: from sqlmodel import Session, create_engine, delete @@ -646,107 +868,125 @@ def _clear_all_from_database(self) -> None: pool_pre_ping=True, ) with Session(engine) as session: - # Delete all analysis results (cascades to Traces and DataAnalysis) + # Drop all runs (cascades to Traces, DataAnalysis, FOVAnalysis via FK) session.exec(delete(CaliResult)) - - # Delete ALL ROIs (cascades to Traces, DataAnalysis, and Masks) - session.exec(delete(ROI)) - - # Delete ALL settings (including orphaned ones from cancelled runs) - session.exec(delete(DetectionSettings)) + # Run-scoped settings always go session.exec(delete(ExtractionSettings)) session.exec(delete(AnalysisSettings)) + session.commit() + # Detection cleanup: per-detection so we can spare kept ones + all_detection_ids = [ + did + for did in session.exec(select(DetectionSettings.id)).all() + if did is not None + ] + for did in all_detection_ids: + if did in keep: + continue + self._delete_detection_data(did, session=session) + + # Untagged ROIs (detection_settings_id is NULL) are unreachable + # without a run, so wipe them too. + session.exec( + delete(ROI).where(ROI.detection_settings_id.is_(None)) # type: ignore[union-attr] + ) session.commit() + self._delete_empty_fovs(session) + engine.dispose(close=True) except Exception as e: QMessageBox.warning(self, "Error", f"Failed to clear all runs: {e}") + def _delete_detection_data( + self, detection_id: int, session: Session | None = None + ) -> None: + """Delete a DetectionSettings row plus all its ROIs/Masks and any empty FOVs. + + If ``session`` is provided, the work happens in that session and the + caller owns the lifecycle. Otherwise a one-off engine/session is opened. + """ + from cali.sqlmodel._model import ROI, Mask + + def _do(s: Session) -> None: + rois = s.exec( + select(ROI).where(ROI.detection_settings_id == detection_id) + ).all() + fov_ids = {roi.fov_id for roi in rois} + mask_ids = {roi.roi_mask_id for roi in rois if roi.roi_mask_id is not None} + + for roi in rois: + s.delete(roi) + s.flush() + + for mask_id in mask_ids: + mask = s.get(Mask, mask_id) + if mask is not None: + s.delete(mask) + + d_settings = s.get(DetectionSettings, detection_id) + if d_settings is not None: + s.delete(d_settings) + + s.commit() + self._delete_empty_fovs(s, fov_ids) + + if session is not None: + _do(session) + return + + if self._database_path is None: + return + try: + from sqlmodel import Session, create_engine + + engine = create_engine(f"sqlite:///{self._database_path}") + with Session(engine) as s: + _do(s) + engine.dispose(close=True) + except Exception as e: + QMessageBox.warning(self, "Error", f"Failed to delete segmentation: {e}") + def _cleanup_orphaned_data( self, session: Session, detection_id: int | None, analysis_id: int | None, + *, + keep_detection: bool = False, ) -> None: - """Clean up orphaned settings and ROIs after deleting a run. + """Clean up settings/ROIs orphaned by a single-run deletion. Parameters ---------- session : Session - Database session - detection_id : int | None - Detection settings ID to check - analysis_id : int | None - Analysis settings ID to check + Active session (caller owns transaction lifecycle). + detection_id, analysis_id : int | None + Settings IDs from the just-deleted run. + keep_detection : bool + If True, never delete the detection (even if orphaned). """ - from cali.sqlmodel._model import ( - FOV, - ROI, - AnalysisSettings, - DetectionSettings, - ) - - # Check if DetectionSettings are orphaned (not used by any other run) - if detection_id is not None: + # Detection + ROIs + if detection_id is not None and not keep_detection: other_runs_using_detection = session.exec( select(CaliResult).where( CaliResult.detection_settings_id == detection_id ) ).first() - if not other_runs_using_detection: - # No other runs use this detection - delete the settings and ROIs cali_logger.info( f"🧹 Cleaning up orphaned DetectionSettings #{detection_id}" ) + self._delete_detection_data(detection_id, session=session) - # Delete all ROIs with this detection_settings_id. These ROIs are - # deleted even if their FOV contains ROIs from other detections - # (This will cascade to delete Traces, DataAnalysis, and Masks) - rois_to_delete = session.exec( - select(ROI).where(ROI.detection_settings_id == detection_id) - ).all() - - roi_count = len(rois_to_delete) - fov_ids = {roi.fov_id for roi in rois_to_delete} - - for roi in rois_to_delete: - session.delete(roi) - - # Delete the detection settings - detection_settings = session.get(DetectionSettings, detection_id) - if detection_settings: - session.delete(detection_settings) - - session.commit() - - cali_logger.info( - f" Deleted {roi_count} ROIs from {len(fov_ids)} FOV(s)" - ) - - # Clean up empty FOVs - # (only delete FOVs that have NO ROIs left from any detection) - for fov_id in fov_ids: - fov = session.get(FOV, fov_id) - if fov: - # Refresh to get updated relationships - session.refresh(fov) - if not fov.rois: - cali_logger.info(f" Deleting empty FOV {fov.name}") - session.delete(fov) - - session.commit() - - # Check if AnalysisSettings are orphaned (not used by any other run) + # Analysis settings if analysis_id is not None: other_runs_using_analysis = session.exec( select(CaliResult).where(CaliResult.analysis_settings_id == analysis_id) ).first() - if not other_runs_using_analysis: - # No other runs use this analysis - delete the settings cali_logger.info( f"🧹 Cleaning up orphaned AnalysisSettings #{analysis_id}" ) @@ -755,41 +995,39 @@ def _cleanup_orphaned_data( session.delete(analysis_settings) session.commit() - def _on_item_clicked(self, item: QListWidgetItem) -> None: - """Handle run item click. + def _delete_empty_fovs( + self, session: Session, fov_ids: set[int] | None = None + ) -> None: + """Delete FOVs that no longer have any ROIs. - Parameters - ---------- - item : QListWidgetItem - The clicked item + If ``fov_ids`` is None, scan every FOV. """ - run_id = item.data(Qt.ItemDataRole.UserRole) - if run_id is not None: - self.runSelected.emit(run_id) + from cali.sqlmodel._model import FOV - def eventFilter(self, a0: QObject | None, a1: QEvent | None) -> bool: - """Filter events to allow deselecting by clicking empty area in list. + if fov_ids is None: + candidates = session.exec(select(FOV)).all() + else: + candidates = [ + fov for fov_id in fov_ids if (fov := session.get(FOV, fov_id)) + ] - Parameters - ---------- - a0 : QObject | None - The object that received the event - a1 : QEvent | None - The event to filter + for fov in candidates: + session.refresh(fov) + if not fov.rois: + cali_logger.info(f" Deleting empty FOV {fov.name}") + session.delete(fov) - Returns - ------- - bool - True if event was handled, False otherwise - """ - if ( - a0 == self._runs_list.viewport() - and a1 - and a1.type() == QEvent.Type.MouseButtonPress - ): - # Check if click is on empty area - item = self._runs_list.itemAt(a1.pos()) - if item is None: - # Clicked on white area - deselect all - self._runs_list.clearSelection() + session.commit() + + # ------------------------------------------------------------------ events + + def eventFilter(self, a0: QObject | None, a1: QEvent | None) -> bool: + """Allow deselecting by clicking empty area in either list.""" + if a1 and a1.type() == QEvent.Type.MouseButtonPress: + if a0 is self._runs_list.viewport(): + if self._runs_list.itemAt(a1.pos()) is None: + self._runs_list.clearSelection() + elif a0 is self._saved_segs_list.viewport(): + if self._saved_segs_list.itemAt(a1.pos()) is None: + self._saved_segs_list.clearSelection() return bool(super().eventFilter(a0, a1)) From ed21f0aa7ffdf335f6f65a89f7be75bd37718510 Mon Sep 17 00:00:00 2001 From: fdrgsp Date: Sun, 17 May 2026 17:38:29 -0400 Subject: [PATCH 2/5] feat: add tooltip for saved segmentations in runs panel --- src/cali/gui/_runs_panel.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/src/cali/gui/_runs_panel.py b/src/cali/gui/_runs_panel.py index 40e84eb6..453d084c 100644 --- a/src/cali/gui/_runs_panel.py +++ b/src/cali/gui/_runs_panel.py @@ -143,13 +143,21 @@ def __init__(self, parent: QWidget | None = None) -> None: saved_layout = QVBoxLayout(saved_container) saved_layout.setContentsMargins(0, 0, 0, 0) saved_layout.setSpacing(2) - saved_layout.addWidget(QLabel("Saved Segmentations")) + _SAVED_SEGS_TOOLTIP = ( + "Each segmentation is normally stored as part of a run.\n" + "When you delete a run, cali asks whether you also want to delete\n" + "the segmentation (ROIs + masks) that was produced by that run.\n" + "If you choose to keep it, it is preserved here — independent of\n" + "any run — so you can inspect its labels or reuse it later.\n\n" + "Click an entry to load its detection settings into the Detection\n" + "tab and preview its labels in the image viewer." + ) + saved_segs_label = QLabel("Saved Segmentations") + saved_segs_label.setToolTip(_SAVED_SEGS_TOOLTIP) + saved_layout.addWidget(saved_segs_label) self._saved_segs_list = QListWidget() self._saved_segs_list.setAlternatingRowColors(True) - self._saved_segs_list.setToolTip( - "Segmentations kept after their runs were deleted. " - "Click to view labels or reuse in a new run." - ) + self._saved_segs_list.setToolTip(_SAVED_SEGS_TOOLTIP) saved_layout.addWidget(self._saved_segs_list) self._splitter.addWidget(saved_container) self._splitter.setStretchFactor(0, 4) From 12561535ffd7ab68f95223b04dee57db24d3ef09 Mon Sep 17 00:00:00 2001 From: fdrgsp Date: Sun, 17 May 2026 17:54:42 -0400 Subject: [PATCH 3/5] test: update --- tests/test_runs_panel.py | 1029 +++++++++++++++++++++++++++++++++++++- 1 file changed, 1024 insertions(+), 5 deletions(-) diff --git a/tests/test_runs_panel.py b/tests/test_runs_panel.py index 375763e2..47bc38d8 100644 --- a/tests/test_runs_panel.py +++ b/tests/test_runs_panel.py @@ -1,4 +1,4 @@ -"""Tests for RunsPanel.get_run_ids() method.""" +"""Tests for _RunsPanel — runs list, saved-segmentations, and keep-detection flows.""" from __future__ import annotations @@ -7,15 +7,16 @@ from typing import TYPE_CHECKING import pytest +from qtpy.QtCore import Qt +from qtpy.QtWidgets import QDialog, QListWidgetItem, QMessageBox -from cali.gui._runs_panel import _RunsPanel +from cali.gui._runs_panel import _DetectionKeepDialog, _DetectionSummary, _RunsPanel +from cali.sqlmodel import CaliResult, DetectionSettings, ExtractionSettings +from cali.sqlmodel._model import SQLModel if TYPE_CHECKING: from pytestqt.qtbot import QtBot - pass -from cali.sqlmodel import CaliResult, DetectionSettings, ExtractionSettings - @pytest.fixture def test_db_path() -> Path: @@ -430,3 +431,1021 @@ def test_both_incomplete_shows_both_asterisks( # Should have asterisks for both assert f"Extraction ID: {extraction_id} ⚠️" in item_text assert f"Analysis ID: {analysis_id} ⚠️" in item_text + + +# ============================================================================ +# Helpers +# ============================================================================ + + +def _make_db(tmp_path: Path) -> Path: + """Create a fresh SQLite database with the cali schema.""" + from sqlmodel import create_engine + + db_path = tmp_path / "test.cali" + engine = create_engine(f"sqlite:///{db_path}") + SQLModel.metadata.create_all(engine) + engine.dispose(close=True) + return db_path + + +def _add_detection_and_run( + session: object, method: str = "cellpose", model_type: str = "cpsam" +) -> tuple[int, int]: + """Add a DetectionSettings + CaliResult pair; return (detection_id, run_id).""" + d = DetectionSettings(method=method, model_type=model_type) + session.add(d) # type: ignore[union-attr] + session.flush() # type: ignore[union-attr] + r = CaliResult(experiment=1, detection_settings_id=d.id, positions_detected=[0]) + session.add(r) # type: ignore[union-attr] + session.flush() # type: ignore[union-attr] + assert d.id is not None and r.id is not None + return d.id, r.id + + +# ============================================================================ +# _DetectionSummary.label() +# ============================================================================ + + +@pytest.mark.parametrize( + "run_count,roi_count,fov_count,model_type,expected_snippets", + [ + (1, 1, 1, "cpsam", ["1 run", "1 ROI", "1 FOV", "cpsam"]), + (2, 5, 3, "cyto3", ["2 runs", "5 ROIs", "3 FOVs", "cyto3"]), + (1, 0, 0, None, ["1 run", "0 ROIs", "0 FOVs"]), + ], +) +def test_detection_summary_label( + run_count: int, + roi_count: int, + fov_count: int, + model_type: str | None, + expected_snippets: list[str], +) -> None: + s = _DetectionSummary( + detection_id=42, + method="cellpose", + model_type=model_type, + run_count=run_count, + roi_count=roi_count, + fov_count=fov_count, + ) + label = s.label() + for snippet in expected_snippets: + assert snippet in label + + +def test_detection_summary_label_no_model_type_omits_slash() -> None: + s = _DetectionSummary(42, "cellpose", None, 1, 0, 0) + label = s.label() + assert "cellpose" in label + # No model_type → no " / " separator + assert " / " not in label + + +# ============================================================================ +# _DetectionKeepDialog +# ============================================================================ + + +def test_detection_keep_dialog_with_summaries(qtbot: QtBot) -> None: + summaries = [ + _DetectionSummary(1, "cellpose", "cpsam", 2, 5, 2), + _DetectionSummary(2, "cellpose", "cyto3", 1, 3, 1), + ] + dialog = _DetectionKeepDialog(summaries) + qtbot.addWidget(dialog) + assert set(dialog._checkboxes) == {1, 2} + assert dialog.kept_detection_ids() == set() # nothing checked by default + + +def test_detection_keep_dialog_no_summaries(qtbot: QtBot) -> None: + dialog = _DetectionKeepDialog([]) + qtbot.addWidget(dialog) + assert dialog._checkboxes == {} + assert dialog.kept_detection_ids() == set() + + +@pytest.mark.parametrize("checked_ids", [{1}, {2}, {1, 2}]) +def test_detection_keep_dialog_kept_ids(qtbot: QtBot, checked_ids: set) -> None: + summaries = [ + _DetectionSummary(1, "cellpose", "cpsam", 2, 5, 2), + _DetectionSummary(2, "cellpose", "cyto3", 1, 3, 1), + ] + dialog = _DetectionKeepDialog(summaries) + qtbot.addWidget(dialog) + for did in checked_ids: + dialog._checkboxes[did].setChecked(True) + assert dialog.kept_detection_ids() == checked_ids + + +# ============================================================================ +# refresh_runs() — saved-segmentations list +# ============================================================================ + + +def test_refresh_runs_orphan_detection_appears_in_saved_list( + tmp_path: Path, runs_panel: _RunsPanel, qtbot: QtBot +) -> None: + """DetectionSettings with no CaliResult should show in the saved-segs list.""" + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + orphan = DetectionSettings(method="cellpose", model_type="cyto3") + session.add(orphan) + session.commit() + orphan_id = orphan.id + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + assert runs_panel._runs_list.count() == 0 + assert runs_panel._saved_segs_list.count() == 1 + item = runs_panel._saved_segs_list.item(0) + assert item is not None + assert item.data(Qt.ItemDataRole.UserRole) == orphan_id + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +def test_refresh_runs_used_detection_not_in_saved_list( + tmp_path: Path, runs_panel: _RunsPanel, qtbot: QtBot +) -> None: + """DetectionSettings referenced by a CaliResult must NOT appear in saved-segs.""" + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + _add_detection_and_run(session) + session.commit() + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + assert runs_panel._runs_list.count() == 1 + assert runs_panel._saved_segs_list.count() == 0 + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +# ============================================================================ +# Signals +# ============================================================================ + + +def test_segmentation_selected_signal_emitted( + tmp_path: Path, runs_panel: _RunsPanel, qtbot: QtBot +) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + orphan = DetectionSettings(method="cellpose", model_type="cyto3") + session.add(orphan) + session.commit() + orphan_id = orphan.id + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + received: list[int] = [] + runs_panel.segmentationSelected.connect(received.append) + + item = runs_panel._saved_segs_list.item(0) + assert item is not None + runs_panel._on_saved_seg_clicked(item) + + assert received == [orphan_id] + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +def test_run_selected_signal_emitted( + tmp_path: Path, runs_panel: _RunsPanel, qtbot: QtBot +) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + _, run_id = _add_detection_and_run(session) + session.commit() + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + received: list[int] = [] + runs_panel.runSelected.connect(received.append) + + item = runs_panel._runs_list.item(0) + assert item is not None + runs_panel._on_run_item_clicked(item) + + assert received == [run_id] + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +# ============================================================================ +# Mutual exclusion of list selections +# ============================================================================ + + +def _make_db_with_run_and_orphan(tmp_path: Path) -> Path: + """DB with one run (detection used) and one orphan detection.""" + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + _add_detection_and_run(session) + orphan = DetectionSettings(method="cellpose", model_type="cyto3") + session.add(orphan) + session.commit() + engine.dispose(close=True) + return db_path + + +def test_selecting_run_clears_saved_segs_selection( + tmp_path: Path, runs_panel: _RunsPanel, qtbot: QtBot +) -> None: + db_path = _make_db_with_run_and_orphan(tmp_path) + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + # Pre-select the saved seg + seg_item = runs_panel._saved_segs_list.item(0) + runs_panel._saved_segs_list.setCurrentItem(seg_item) + qtbot.wait(20) + assert runs_panel._saved_segs_list.selectedItems() + + # Select the run + run_item = runs_panel._runs_list.item(0) + runs_panel._runs_list.setCurrentItem(run_item) + qtbot.wait(20) + + assert not runs_panel._saved_segs_list.selectedItems() + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +def test_selecting_saved_seg_clears_run_selection( + tmp_path: Path, runs_panel: _RunsPanel, qtbot: QtBot +) -> None: + db_path = _make_db_with_run_and_orphan(tmp_path) + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + # Pre-select the run + run_item = runs_panel._runs_list.item(0) + runs_panel._runs_list.setCurrentItem(run_item) + qtbot.wait(20) + assert runs_panel._runs_list.selectedItems() + + # Select the saved seg + seg_item = runs_panel._saved_segs_list.item(0) + runs_panel._saved_segs_list.setCurrentItem(seg_item) + qtbot.wait(20) + + assert not runs_panel._runs_list.selectedItems() + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +# ============================================================================ +# Delete button state +# ============================================================================ + + +@pytest.mark.parametrize( + "select_run,select_seg,expect_enabled", + [ + (True, False, True), + (False, True, True), + (False, False, False), + ], +) +def test_delete_button_state( + tmp_path: Path, + runs_panel: _RunsPanel, + qtbot: QtBot, + select_run: bool, + select_seg: bool, + expect_enabled: bool, +) -> None: + db_path = _make_db_with_run_and_orphan(tmp_path) + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + runs_panel._runs_list.clearSelection() + runs_panel._saved_segs_list.clearSelection() + runs_panel._delete_btn.setEnabled(False) + + if select_run: + runs_panel._runs_list.setCurrentItem(runs_panel._runs_list.item(0)) + if select_seg: + runs_panel._saved_segs_list.setCurrentItem(runs_panel._saved_segs_list.item(0)) + + runs_panel._update_delete_button() + assert runs_panel._delete_btn.isEnabled() is expect_enabled + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +# ============================================================================ +# get_selected_saved_segmentation_id() / get_selected_detection_settings_id() +# ============================================================================ + + +def test_get_selected_saved_segmentation_id_none_when_nothing_selected( + runs_panel: _RunsPanel, +) -> None: + assert runs_panel.get_selected_saved_segmentation_id() is None + + +def test_get_selected_saved_segmentation_id_returns_id( + tmp_path: Path, runs_panel: _RunsPanel, qtbot: QtBot +) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + orphan = DetectionSettings(method="cellpose", model_type="cyto3") + session.add(orphan) + session.commit() + orphan_id = orphan.id + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + runs_panel._saved_segs_list.setCurrentItem(runs_panel._saved_segs_list.item(0)) + + assert runs_panel.get_selected_saved_segmentation_id() == orphan_id + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +def test_get_selected_detection_settings_id_none_when_nothing_selected( + runs_panel: _RunsPanel, +) -> None: + assert runs_panel.get_selected_detection_settings_id() is None + + +def test_get_selected_detection_settings_id_from_saved_seg( + tmp_path: Path, runs_panel: _RunsPanel, qtbot: QtBot +) -> None: + """When no run is selected, returns the saved seg's DetectionSettings ID.""" + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + orphan = DetectionSettings(method="cellpose", model_type="cyto3") + session.add(orphan) + session.commit() + orphan_id = orphan.id + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + runs_panel._runs_list.clearSelection() + runs_panel._saved_segs_list.setCurrentItem(runs_panel._saved_segs_list.item(0)) + + assert runs_panel.get_selected_detection_settings_id() == orphan_id + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +# ============================================================================ +# Internal DB helpers +# ============================================================================ + + +def test_count_runs_using_detection_no_database(runs_panel: _RunsPanel) -> None: + assert runs_panel._count_runs_using_detection(1) == 0 + + +def test_count_runs_using_detection(tmp_path: Path, runs_panel: _RunsPanel) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + d = DetectionSettings(method="cellpose", model_type="cpsam") + session.add(d) + session.flush() + for _ in range(2): + session.add( + CaliResult( + experiment=1, + detection_settings_id=d.id, + positions_detected=[0], + ) + ) + session.commit() + detection_id = d.id + engine.dispose(close=True) + + runs_panel._database_path = db_path + assert runs_panel._count_runs_using_detection(detection_id) == 2 + + +def test_get_detection_id_for_run_no_database(runs_panel: _RunsPanel) -> None: + assert runs_panel._get_detection_id_for_run(999) is None + + +def test_get_detection_id_for_run(tmp_path: Path, runs_panel: _RunsPanel) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + detection_id, run_id = _add_detection_and_run(session) + session.commit() + engine.dispose(close=True) + + runs_panel._database_path = db_path + assert runs_panel._get_detection_id_for_run(run_id) == detection_id + + +# ============================================================================ +# _delete_run_from_database() — keep_detection flag +# ============================================================================ + + +def test_delete_run_keep_detection_preserves_detection( + tmp_path: Path, runs_panel: _RunsPanel +) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + detection_id, run_id = _add_detection_and_run(session) + session.commit() + engine.dispose(close=True) + + runs_panel._database_path = db_path + runs_panel._delete_run_from_database(run_id, keep_detection=True) + + engine2 = create_engine(f"sqlite:///{db_path}") + with Session(engine2) as session: + assert session.get(CaliResult, run_id) is None + assert session.get(DetectionSettings, detection_id) is not None + engine2.dispose(close=True) + + +def test_delete_run_without_keep_detection_removes_detection( + tmp_path: Path, runs_panel: _RunsPanel +) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + detection_id, run_id = _add_detection_and_run(session) + session.commit() + engine.dispose(close=True) + + runs_panel._database_path = db_path + runs_panel._delete_run_from_database(run_id, keep_detection=False) + + engine2 = create_engine(f"sqlite:///{db_path}") + with Session(engine2) as session: + assert session.get(CaliResult, run_id) is None + assert session.get(DetectionSettings, detection_id) is None + engine2.dispose(close=True) + + +# ============================================================================ +# _delete_detection_data() +# ============================================================================ + + +def test_delete_detection_data_removes_detection_rois_and_empty_fov( + tmp_path: Path, runs_panel: _RunsPanel +) -> None: + from sqlmodel import Session, create_engine + + from cali.sqlmodel._model import FOV, ROI + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + fov = FOV(name="A1_0000", position_index=0) + session.add(fov) + session.flush() + d = DetectionSettings(method="cellpose", model_type="cpsam") + session.add(d) + session.flush() + roi = ROI(fov_id=fov.id, label_value=1, detection_settings_id=d.id) + session.add(roi) + session.commit() + detection_id = d.id + fov_id = fov.id + engine.dispose(close=True) + + runs_panel._database_path = db_path + runs_panel._delete_detection_data(detection_id) + + from sqlmodel import Session, create_engine, select + + engine2 = create_engine(f"sqlite:///{db_path}") + with Session(engine2) as session: + assert session.get(DetectionSettings, detection_id) is None + assert ( + session.exec( + select(ROI).where(ROI.detection_settings_id == detection_id) + ).all() + == [] + ) + # FOV becomes empty → should be removed + assert session.get(FOV, fov_id) is None + engine2.dispose(close=True) + + +# ============================================================================ +# _clear_all_from_database() +# ============================================================================ + + +def test_clear_all_from_database_keeps_specified_detection( + tmp_path: Path, runs_panel: _RunsPanel +) -> None: + from sqlmodel import Session, create_engine, select + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + keep_id, _ = _add_detection_and_run(session, model_type="cpsam") + delete_id, _ = _add_detection_and_run(session, model_type="cyto3") + session.commit() + engine.dispose(close=True) + + runs_panel._database_path = db_path + runs_panel._clear_all_from_database(keep_detection_ids={keep_id}) + + engine2 = create_engine(f"sqlite:///{db_path}") + with Session(engine2) as session: + assert session.get(DetectionSettings, keep_id) is not None + assert session.get(DetectionSettings, delete_id) is None + assert session.exec(select(CaliResult)).all() == [] + engine2.dispose(close=True) + + +def test_clear_all_from_database_deletes_everything_by_default( + tmp_path: Path, runs_panel: _RunsPanel +) -> None: + from sqlmodel import Session, create_engine, select + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + detection_id, _ = _add_detection_and_run(session) + session.commit() + engine.dispose(close=True) + + runs_panel._database_path = db_path + runs_panel._clear_all_from_database() + + engine2 = create_engine(f"sqlite:///{db_path}") + with Session(engine2) as session: + assert session.get(DetectionSettings, detection_id) is None + assert session.exec(select(CaliResult)).all() == [] + engine2.dispose(close=True) + + +# ============================================================================ +# _delete_selected_saved_segmentation() +# ============================================================================ + + +def test_delete_selected_saved_segmentation_confirmed( + tmp_path: Path, + runs_panel: _RunsPanel, + qtbot: QtBot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + orphan = DetectionSettings(method="cellpose", model_type="cyto3") + session.add(orphan) + session.commit() + orphan_id = orphan.id + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + runs_panel._saved_segs_list.setCurrentItem(runs_panel._saved_segs_list.item(0)) + monkeypatch.setattr( + QMessageBox, "warning", lambda *a, **kw: QMessageBox.StandardButton.Yes + ) + + emitted: list[bool] = [] + runs_panel.settingsDeleted.connect(lambda: emitted.append(True)) + runs_panel._delete_selected_saved_segmentation() + qtbot.wait(50) + + assert runs_panel._saved_segs_list.count() == 0 + assert emitted + + engine2 = create_engine(f"sqlite:///{db_path}") + with Session(engine2) as session: + assert session.get(DetectionSettings, orphan_id) is None + engine2.dispose(close=True) + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +def test_delete_selected_saved_segmentation_cancelled( + tmp_path: Path, + runs_panel: _RunsPanel, + qtbot: QtBot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + orphan = DetectionSettings(method="cellpose", model_type="cyto3") + session.add(orphan) + session.commit() + orphan_id = orphan.id + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + runs_panel._saved_segs_list.setCurrentItem(runs_panel._saved_segs_list.item(0)) + monkeypatch.setattr( + QMessageBox, "warning", lambda *a, **kw: QMessageBox.StandardButton.No + ) + + runs_panel._delete_selected_saved_segmentation() + qtbot.wait(50) + + assert runs_panel._saved_segs_list.count() == 1 + + engine2 = create_engine(f"sqlite:///{db_path}") + with Session(engine2) as session: + assert session.get(DetectionSettings, orphan_id) is not None + engine2.dispose(close=True) + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +# ============================================================================ +# _delete_selected_run() — sole detection keep/delete/cancel +# ============================================================================ + + +@pytest.mark.parametrize( + "keep_choice,detection_survives", + [(True, True), (False, False)], +) +def test_delete_selected_run_sole_detection( + tmp_path: Path, + runs_panel: _RunsPanel, + qtbot: QtBot, + monkeypatch: pytest.MonkeyPatch, + keep_choice: bool, + detection_survives: bool, +) -> None: + """When run is sole user of detection, keep_choice determines if detection survives.""" # noqa: E501 + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + detection_id, run_id = _add_detection_and_run(session) + session.commit() + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + runs_panel._runs_list.setCurrentItem(runs_panel._runs_list.item(0)) + monkeypatch.setattr( + runs_panel, "_ask_keep_or_delete_segmentation", lambda *a: keep_choice + ) + + runs_panel._delete_selected_run() + qtbot.wait(50) + + assert runs_panel._runs_list.count() == 0 + + engine2 = create_engine(f"sqlite:///{db_path}") + with Session(engine2) as session: + assert session.get(CaliResult, run_id) is None + det = session.get(DetectionSettings, detection_id) + assert (det is not None) is detection_survives + engine2.dispose(close=True) + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +def test_delete_selected_run_sole_detection_cancel( + tmp_path: Path, + runs_panel: _RunsPanel, + qtbot: QtBot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + _, _run_id = _add_detection_and_run(session) + session.commit() + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + runs_panel._runs_list.setCurrentItem(runs_panel._runs_list.item(0)) + monkeypatch.setattr(runs_panel, "_ask_keep_or_delete_segmentation", lambda *a: None) + + runs_panel._delete_selected_run() + qtbot.wait(50) + + # Nothing deleted after cancel + assert runs_panel._runs_list.count() == 1 + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +def test_delete_selected_run_shared_detection_shows_simple_confirm( + tmp_path: Path, + runs_panel: _RunsPanel, + qtbot: QtBot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Shared detection → simple Yes/No QMessageBox (no keep dialog).""" + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + d = DetectionSettings(method="cellpose", model_type="cpsam") + session.add(d) + session.flush() + r1 = CaliResult( + experiment=1, detection_settings_id=d.id, positions_detected=[0] + ) + r2 = CaliResult( + experiment=1, detection_settings_id=d.id, positions_detected=[1] + ) + session.add_all([r1, r2]) + session.commit() + detection_id = d.id + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + runs_panel._runs_list.setCurrentItem(runs_panel._runs_list.item(0)) + monkeypatch.setattr( + QMessageBox, "warning", lambda *a, **kw: QMessageBox.StandardButton.Yes + ) + + runs_panel._delete_selected_run() + qtbot.wait(50) + + # One run deleted, other remains; shared detection must still exist + assert runs_panel._runs_list.count() == 1 + engine2 = create_engine(f"sqlite:///{db_path}") + with Session(engine2) as session: + assert session.get(DetectionSettings, detection_id) is not None + engine2.dispose(close=True) + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +# ============================================================================ +# _delete_selected() routing +# ============================================================================ + + +def test_delete_selected_routes_to_saved_seg_handler( + runs_panel: _RunsPanel, monkeypatch: pytest.MonkeyPatch +) -> None: + called: list[str] = [] + monkeypatch.setattr( + runs_panel, "_delete_selected_saved_segmentation", lambda: called.append("seg") + ) + monkeypatch.setattr( + runs_panel, "_delete_selected_run", lambda: called.append("run") + ) + + item = QListWidgetItem("test-seg") + runs_panel._saved_segs_list.addItem(item) + runs_panel._saved_segs_list.setCurrentItem(item) + + runs_panel._delete_selected() + assert called == ["seg"] + + +def test_delete_selected_routes_to_run_handler( + runs_panel: _RunsPanel, monkeypatch: pytest.MonkeyPatch +) -> None: + called: list[str] = [] + monkeypatch.setattr( + runs_panel, "_delete_selected_saved_segmentation", lambda: called.append("seg") + ) + monkeypatch.setattr( + runs_panel, "_delete_selected_run", lambda: called.append("run") + ) + + # No saved-seg selected → goes to run handler + runs_panel._delete_selected() + assert called == ["run"] + + +# ============================================================================ +# _build_summary() / _all_detection_summaries() +# ============================================================================ + + +def test_build_summary_counts(tmp_path: Path, runs_panel: _RunsPanel) -> None: + from sqlmodel import Session, create_engine + + from cali.sqlmodel._model import FOV, ROI + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + fov1 = FOV(name="A1_0000", position_index=0) + fov2 = FOV(name="A1_0001", position_index=1) + session.add_all([fov1, fov2]) + session.flush() + + d = DetectionSettings(method="cellpose", model_type="cpsam") + session.add(d) + session.flush() + + session.add( + CaliResult(experiment=1, detection_settings_id=d.id, positions_detected=[0]) + ) + + for fov_id, n_rois in [(fov1.id, 2), (fov2.id, 1)]: + for label in range(1, n_rois + 1): + session.add( + ROI(fov_id=fov_id, label_value=label, detection_settings_id=d.id) + ) + + session.commit() + d_settings = session.get(DetectionSettings, d.id) + summary = runs_panel._build_summary(session, d_settings) + + engine.dispose(close=True) + + assert summary.detection_id == d.id + assert summary.run_count == 1 + assert summary.roi_count == 3 + assert summary.fov_count == 2 + + +def test_all_detection_summaries_no_database(runs_panel: _RunsPanel) -> None: + assert runs_panel._all_detection_summaries() == [] + + +# ============================================================================ +# _clear_all_runs() — dialog interactions +# ============================================================================ + + +def test_clear_all_runs_cancelled_does_nothing( + tmp_path: Path, + runs_panel: _RunsPanel, + qtbot: QtBot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + _add_detection_and_run(session) + session.commit() + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + monkeypatch.setattr( + _DetectionKeepDialog, "exec", lambda self: QDialog.DialogCode.Rejected + ) + + runs_panel._clear_all_runs() + qtbot.wait(50) + + assert runs_panel._runs_list.count() == 1 + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() + + +def test_clear_all_runs_accepted_with_kept_detection( + tmp_path: Path, + runs_panel: _RunsPanel, + qtbot: QtBot, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from sqlmodel import Session, create_engine + + db_path = _make_db(tmp_path) + engine = create_engine(f"sqlite:///{db_path}") + with Session(engine) as session: + keep_id, _ = _add_detection_and_run(session, model_type="cpsam") + delete_id, _ = _add_detection_and_run(session, model_type="cyto3") + session.commit() + engine.dispose(close=True) + + runs_panel.set_database_path(db_path) + qtbot.wait(50) + + monkeypatch.setattr( + _DetectionKeepDialog, "exec", lambda self: QDialog.DialogCode.Accepted + ) + monkeypatch.setattr( + _DetectionKeepDialog, "kept_detection_ids", lambda self: {keep_id} + ) + + runs_panel._clear_all_runs() + qtbot.wait(50) + + # All runs gone; kept detection becomes a saved seg + assert runs_panel._runs_list.count() == 0 + assert runs_panel._saved_segs_list.count() == 1 + + engine2 = create_engine(f"sqlite:///{db_path}") + with Session(engine2) as session: + assert session.get(DetectionSettings, keep_id) is not None + assert session.get(DetectionSettings, delete_id) is None + engine2.dispose(close=True) + + runs_panel.clear() + runs_panel._database_path = None + qtbot.wait(50) + gc.collect() From 2e3388d0b5ced7b8b1236bca0b67a1b8860c6a1e Mon Sep 17 00:00:00 2001 From: fdrgsp Date: Sun, 17 May 2026 18:41:47 -0400 Subject: [PATCH 4/5] fix --- src/cali/gui/_cali_gui.py | 3 +- src/cali/gui/_runs_panel.py | 58 ++++++++++++++++++++----------------- 2 files changed, 33 insertions(+), 28 deletions(-) diff --git a/src/cali/gui/_cali_gui.py b/src/cali/gui/_cali_gui.py index 61f891f5..cb215999 100644 --- a/src/cali/gui/_cali_gui.py +++ b/src/cali/gui/_cali_gui.py @@ -2436,7 +2436,8 @@ def _on_saved_segmentation_selected(self, detection_settings_id: int) -> None: d_settings = DetectionSettings.load_from_database( self._database_path, id=detection_settings_id ) - assert isinstance(d_settings, DetectionSettings) + if not isinstance(d_settings, DetectionSettings): # pragma: no cover + raise TypeError(f"Expected DetectionSettings, got {type(d_settings)}") if d_settings.method == "cellpose": self._detection_wdg.setValue( CellposeSettingsData( diff --git a/src/cali/gui/_runs_panel.py b/src/cali/gui/_runs_panel.py index 453d084c..a4e5668b 100644 --- a/src/cali/gui/_runs_panel.py +++ b/src/cali/gui/_runs_panel.py @@ -66,13 +66,13 @@ def __init__( self, summaries: list[_DetectionSummary], parent: QWidget | None = None ) -> None: super().__init__(parent) - self.setWindowTitle("Delete All Runs") + self.setWindowTitle("Delete All") layout = QVBoxLayout(self) layout.addWidget( QLabel( "All runs will be deleted.\n" - "Tick any segmentations you want to keep — unticked ones will be " - "deleted along with their ROIs." + "Tick any segmentations you want to keep — unticked ones will also " + "be deleted along with their ROIs." ) ) self._checkboxes: dict[int, QCheckBox] = {} @@ -173,18 +173,21 @@ def __init__(self, parent: QWidget | None = None) -> None: self._delete_btn = QPushButton("Delete Selected") self._delete_btn.setIcon(QIconifyIcon("mdi:delete", color=RED)) self._delete_btn.setToolTip( - "Delete the selected run or saved segmentation from the database" + "Delete the selected run (if it's the only run left, you will be asked if " + "you want to keep the segmentations, if any)." ) self._delete_btn.clicked.connect(self._delete_selected) self._delete_btn.setEnabled(False) buttons_layout.addWidget(self._delete_btn) # Clear all button - self._clear_all_btn = QPushButton("Delete All") - self._clear_all_btn.setIcon(QIconifyIcon("mdi:delete-forever", color=RED)) - self._clear_all_btn.setToolTip("Delete all runs from the database") - self._clear_all_btn.clicked.connect(self._clear_all_runs) - buttons_layout.addWidget(self._clear_all_btn) + self._delete_all_btn = QPushButton("Delete All") + self._delete_all_btn.setIcon(QIconifyIcon("mdi:delete-forever", color=RED)) + self._delete_all_btn.setToolTip( + "Delete all runs (you will be asked if you want to keep any segmentations)." + ) + self._delete_all_btn.clicked.connect(self._clear_all_runs) + buttons_layout.addWidget(self._delete_all_btn) layout.addLayout(buttons_layout) @@ -823,27 +826,28 @@ def _delete_run_from_database( from sqlmodel import Session, create_engine engine = create_engine(f"sqlite:///{self._database_path}") - with Session(engine) as session: - result = session.get(CaliResult, run_id) - if not result: - return + try: + with Session(engine) as session: + result = session.get(CaliResult, run_id) + if not result: + return - detection_id = result.detection_settings_id - analysis_id = result.analysis_settings_id + detection_id = result.detection_settings_id + analysis_id = result.analysis_settings_id - # Delete the analysis result (cascades to Traces via FK) - session.delete(result) - session.commit() - - # Clean up orphaned settings (and ROIs unless keep_detection) - self._cleanup_orphaned_data( - session, - detection_id, - analysis_id, - keep_detection=keep_detection, - ) + # Delete the analysis result (cascades to Traces via FK) + session.delete(result) + session.commit() - engine.dispose(close=True) + # Clean up orphaned settings (and ROIs unless keep_detection) + self._cleanup_orphaned_data( + session, + detection_id, + analysis_id, + keep_detection=keep_detection, + ) + finally: + engine.dispose(close=True) except Exception as e: QMessageBox.warning(self, "Error", f"Failed to delete run: {e}") From 10128d36d3cad7d28e8836c530faa35bfbbc0cec Mon Sep 17 00:00:00 2001 From: fdrgsp Date: Sun, 17 May 2026 18:42:11 -0400 Subject: [PATCH 5/5] fix --- src/cali/gui/_runs_panel.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/cali/gui/_runs_panel.py b/src/cali/gui/_runs_panel.py index a4e5668b..42b2ef36 100644 --- a/src/cali/gui/_runs_panel.py +++ b/src/cali/gui/_runs_panel.py @@ -174,7 +174,7 @@ def __init__(self, parent: QWidget | None = None) -> None: self._delete_btn.setIcon(QIconifyIcon("mdi:delete", color=RED)) self._delete_btn.setToolTip( "Delete the selected run (if it's the only run left, you will be asked if " - "you want to keep the segmentations, if any)." + "you want to keep the segmentation)." ) self._delete_btn.clicked.connect(self._delete_selected) self._delete_btn.setEnabled(False)