From d2cc7777d61c265520bfacee0e539ec197e59c6d Mon Sep 17 00:00:00 2001 From: Eitan Yarmush Date: Wed, 15 Jul 2026 12:02:27 -0400 Subject: [PATCH] feat: add roll-aware auto crop --- negpy/desktop/controller.py | 134 ++++- negpy/desktop/session.py | 4 + negpy/desktop/view/sidebar/geometry.py | 12 + negpy/desktop/workers/render.py | 119 +++++ negpy/features/geometry/batch_autocrop.py | 564 ++++++++++++++++++++++ tests/test_batch_autocrop.py | 163 +++++++ tests/test_batch_progress.py | 119 ++++- tests/test_controller.py | 89 ++++ 8 files changed, 1201 insertions(+), 3 deletions(-) create mode 100644 negpy/features/geometry/batch_autocrop.py create mode 100644 tests/test_batch_autocrop.py diff --git a/negpy/desktop/controller.py b/negpy/desktop/controller.py index 9328bb17..d6cf7b28 100644 --- a/negpy/desktop/controller.py +++ b/negpy/desktop/controller.py @@ -14,6 +14,8 @@ from negpy.desktop.workers.render import ( AssetDiscoveryTask, AssetDiscoveryWorker, + AutocropBatchTask, + AutocropBatchWorker, NormalizationTask, NormalizationWorker, PreviewLoadTask, @@ -133,6 +135,7 @@ class AppController(QObject): render_requested = pyqtSignal(RenderTask) preview_load_requested = pyqtSignal(PreviewLoadTask) normalization_requested = pyqtSignal(NormalizationTask) + autocrop_batch_requested = pyqtSignal(AutocropBatchTask) analysis_buffer_preview_requested = pyqtSignal(float) rotation_guide_requested = pyqtSignal() crop_guide_changed = pyqtSignal() @@ -229,6 +232,8 @@ def __init__(self, session_manager: DesktopSessionManager): self.norm_thread = QThread() self.norm_worker = NormalizationWorker(self.preview_service, self.session.repo) self.norm_worker.moveToThread(self.norm_thread) + self.autocrop_batch_worker = AutocropBatchWorker(self.preview_service, self.session.repo) + self.autocrop_batch_worker.moveToThread(self.norm_thread) self.norm_thread.start() self.discovery_thread = QThread() @@ -390,6 +395,13 @@ def _connect_signals(self) -> None: self.norm_worker.error.connect(self._on_render_error) self.norm_worker.error.connect(self._on_batch_error) + self.autocrop_batch_requested.connect(self.autocrop_batch_worker.process) + self.autocrop_batch_worker.progress.connect(self._on_autocrop_batch_progress) + self.autocrop_batch_worker.finished.connect(self._on_autocrop_batch_finished) + self.autocrop_batch_worker.cancelled.connect(self._on_batch_cancelled) + self.autocrop_batch_worker.error.connect(self._on_render_error) + self.autocrop_batch_worker.error.connect(self._on_batch_error) + self.asset_discovery_requested.connect(self.discovery_worker.process) self.discovery_worker.progress.connect(self._on_discovery_progress) self.discovery_worker.finished.connect(self._on_discovery_finished) @@ -435,6 +447,8 @@ def _connect_signals(self) -> None: self.session.files_changed.connect(self._render_debounce.start) def generate_missing_thumbnails(self) -> None: + if self._active_batch is not None: + return missing = [f for f in self.state.uploaded_files if f["name"] not in self.state.thumbnails] if missing: self.set_status("GENERATING THUMBNAILS...") @@ -459,7 +473,7 @@ def _on_thumbnails_finished(self, new_thumbs: Dict[str, Any]) -> None: # --- Batch progress popup ------------------------------------------------- def _begin_batch(self, title: str, abortable: bool) -> None: - self._active_batch = title if abortable else None + self._active_batch = title self.batch_started.emit(title, abortable) def _end_batch(self) -> None: @@ -482,6 +496,8 @@ def abort_active_batch(self) -> None: self.export_worker.cancel() elif self._active_batch == "Analyzing roll": self.norm_worker.cancel() + elif self._active_batch == "Auto cropping roll": + self.autocrop_batch_worker.cancel() def saved_session_paths(self) -> List[str]: """Returns last session's file paths that still exist on disk.""" @@ -514,6 +530,9 @@ def request_asset_discovery( appending) and reselects `reselect_path` — used when re-running discovery over already-loaded files (e.g. an RGB-scan mode toggle). """ + if self._active_batch is not None and not self._discovery_running: + self.set_status(f"Wait for {self._active_batch.lower()} to finish", 3000) + return request = _DiscoveryRequest( paths=tuple(paths), auto_open=auto_open, @@ -942,6 +961,111 @@ def apply_auto_crop(self) -> None: ) self.request_render() + def apply_auto_crop_all(self) -> None: + """Detect, calibrate, and persist crops for every visible uncropped frame.""" + if self._active_batch is not None: + return + visible_files = [self.state.uploaded_files[i] for i in self.session.asset_model.visible_actual_indices_ordered()] + if not visible_files: + return + + ratio_str = self.state.config.geometry.autocrop_ratio + try: + first, second = (float(part) for part in ratio_str.split(":", 1)) + frame_ratio = max(first, second) / min(first, second) + except (TypeError, ValueError, ZeroDivisionError): + frame_ratio = 1.5 + + manual_count = 0 + for file_info in visible_files: + config = self.session.repo.load_file_settings(file_info["hash"]) + if config and config.geometry.manual_crop_rect is not None: + manual_count += 1 + + reply = QMessageBox.question( + None, + "Auto Crop All", + f"Analyze {len(visible_files)} visible frame(s) together using the {ratio_str} ratio?\n\n" + "Confident frames calibrate ambiguous ones. Existing manual crops will be " + f"kept ({manual_count}), and unsupported frames will remain unchanged.", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.Cancel, + QMessageBox.StandardButton.Yes, + ) + if reply != QMessageBox.StandardButton.Yes: + return + + self.set_status("Starting Auto Crop All...") + self._begin_batch("Auto cropping roll", abortable=True) + self.autocrop_batch_requested.emit( + AutocropBatchTask( + files=visible_files, + workspace_color_space=self.state.workspace_color_space, + default_config=self.session.fresh_file_config(), + frame_ratio=frame_ratio, + ) + ) + + def _on_autocrop_batch_progress(self, current: int, total: int, label: str) -> None: + self.set_status(f"Auto cropping {current}/{total}: {label}...") + self.status_progress_requested.emit(current, total) + self.batch_progress.emit(current, total, label) + + def _on_autocrop_batch_finished(self, payload: dict) -> None: + from negpy.features.geometry.batch_autocrop import autocrop_input_signature, geometry_from_autocrop_result + + self._end_batch() + results = payload["results"] + canvases = payload["canvases"] + applied = 0 + unchanged = len(payload["failed"]) + skipped = len(payload["skipped"]) + current_config = None + + for file_info in self.state.uploaded_files: + source_hash = file_info["hash"] + result = results.get(source_hash) + if result is None: + continue + if result.method == "manual": + unchanged += 1 + continue + + captured_config = payload["configs"][source_hash] + latest_config = self.session.repo.load_file_settings(source_hash) + config = resolve_asset_rgbscan(latest_config, file_info) if latest_config is not None else captured_config + if config.geometry.manual_crop_rect is not None: + skipped += 1 + continue + if autocrop_input_signature(config) != autocrop_input_signature(captured_config): + unchanged += 1 + continue + new_geometry = geometry_from_autocrop_result( + config.geometry, + result, + canvases[source_hash], + ) + if new_geometry is None: + unchanged += 1 + continue + + new_process = replace(config.process, **invalidate_local_bounds(config.process)) + updated = replace(config, geometry=new_geometry, process=new_process) + self.session.repo.save_file_settings(source_hash, updated) + if source_hash == self.state.current_file_hash: + current_config = updated + applied += 1 + + if current_config is not None: + self.session.update_config(current_config, persist=False, render=True, record_history=False) + + self.status_progress_requested.emit(0, 0) + self.set_status( + f"Auto Crop All: cropped {applied}, kept {skipped} manual, left {unchanged} unchanged", + 5000, + ) + if current_config is not None: + self.request_render() + def detect_aspect_ratio(self) -> None: img = self.state.preview_raw if img is None: @@ -1276,6 +1400,8 @@ def request_batch_normalization(self) -> None: """ Initiates background analysis for batch normalization. """ + if self._active_batch is not None: + return visible_files = [self.state.uploaded_files[i] for i in self.session.asset_model.visible_actual_indices_ordered()] if not visible_files: return @@ -2129,6 +2255,8 @@ def _contact_sheet_output_dir(self, visible_files: list) -> Optional[str]: def request_contact_sheet(self) -> None: """Renders all visible files small and writes darkroom contact sheet(s).""" + if self._active_batch is not None: + return visible_files = [self.state.uploaded_files[i] for i in self.session.asset_model.visible_actual_indices_ordered()] if not visible_files: return @@ -2191,6 +2319,8 @@ def export_edit_sidecars(self) -> None: self.set_status(f"Wrote {written} edit sidecar(s)", 4000) def _run_export_tasks(self, tasks: List[ExportTask]) -> None: + if self._active_batch is not None: + return # Reject unencodable format/colour-space pairings before anything else. blocked = [t for t in tasks if export_blocked(t.export_settings.export_fmt, t.export_settings.export_color_space)] if blocked: @@ -2437,6 +2567,8 @@ def cleanup(self) -> None: if self.thumb_thread.isRunning(): self.thumb_thread.quit() self.thumb_thread.wait() + self.norm_worker.cancel() + self.autocrop_batch_worker.cancel() if self.norm_thread.isRunning(): self.norm_thread.quit() self.norm_thread.wait() diff --git a/negpy/desktop/session.py b/negpy/desktop/session.py index ebab79c7..3887b8ce 100644 --- a/negpy/desktop/session.py +++ b/negpy/desktop/session.py @@ -644,6 +644,10 @@ def _apply_sticky_settings(self, config: WorkspaceConfig, only_global: bool = Fa return config + def fresh_file_config(self) -> WorkspaceConfig: + """Build the same clean, sticky-aware config used for a newly opened file.""" + return self._apply_sticky_settings(WorkspaceConfig(), only_global=False) + def _persist_sticky_settings(self, config: WorkspaceConfig) -> None: """ Saves current settings to global storage in a single transaction. diff --git a/negpy/desktop/view/sidebar/geometry.py b/negpy/desktop/view/sidebar/geometry.py index 51ebad48..bf702161 100644 --- a/negpy/desktop/view/sidebar/geometry.py +++ b/negpy/desktop/view/sidebar/geometry.py @@ -116,6 +116,15 @@ def _init_ui(self) -> None: auto_row.addWidget(self.mode_combo, 1) self.layout.addLayout(auto_row) + self.auto_crop_all_btn = QPushButton(" Auto Crop All") + self.auto_crop_all_btn.setIcon(qta.icon("fa5s.layer-group", color=THEME.text_primary)) + self.auto_crop_all_btn.setToolTip( + "Analyze all visible frames together, keep existing manual crops, " + "and save confident results before Batch Analysis (Image only mode)" + ) + self.auto_crop_all_btn.setEnabled(conf.autocrop_mode == AutocropMode.IMAGE) + self.layout.addWidget(self.auto_crop_all_btn) + self.offset_slider = CompactSlider( "Crop Offset", -5.0, @@ -173,6 +182,7 @@ def _connect_signals(self) -> None: self.manual_crop_btn.toggled.connect(self._on_manual_crop_toggled) self.clear_crop_btn.clicked.connect(self.controller.reset_crop) self.reset_crop_btn.toggled.connect(self._on_auto_crop_toggled) + self.auto_crop_all_btn.clicked.connect(self.controller.apply_auto_crop_all) self.offset_slider.valueChanged.connect( lambda v: self.update_config_section("geometry", render=True, persist=False, readback_metrics=False, autocrop_offset=int(v)) @@ -247,6 +257,7 @@ def sync_ui(self) -> None: self.reset_crop_btn.setChecked(conf.auto_crop_enabled) self.manual_crop_btn.set_crop_active(conf.manual_crop_rect is not None) self.reset_crop_btn.set_crop_active(conf.auto_crop_enabled) + self.auto_crop_all_btn.setEnabled(conf.autocrop_mode == AutocropMode.IMAGE) finally: self.block_signals(False) @@ -261,3 +272,4 @@ def block_signals(self, blocked: bool) -> None: self.manual_crop_btn.blockSignals(blocked) self.straighten_btn.blockSignals(blocked) self.reset_crop_btn.blockSignals(blocked) + self.auto_crop_all_btn.blockSignals(blocked) diff --git a/negpy/desktop/workers/render.py b/negpy/desktop/workers/render.py index c736d27b..ffdb062d 100644 --- a/negpy/desktop/workers/render.py +++ b/negpy/desktop/workers/render.py @@ -67,6 +67,16 @@ class NormalizationTask: override_crosstalk_matrix: tuple | None = None +@dataclass(frozen=True) +class AutocropBatchTask: + """Request to detect and calibrate crops across the visible roll.""" + + files: list[dict] + workspace_color_space: str + default_config: WorkspaceConfig + frame_ratio: float + + @dataclass(frozen=True) class AssetDiscoveryTask: """Request to find and hash image files in paths.""" @@ -599,3 +609,112 @@ def get_robust_mean(data: np.ndarray) -> np.ndarray: except Exception as e: logger.error(f"Batch Normalization failure: {e}") self.error.emit(str(e)) + + +class AutocropBatchWorker(QObject): + """Detects crop evidence for a roll and resolves it as one calibrated batch.""" + + progress = pyqtSignal(int, int, str) + finished = pyqtSignal(dict) + cancelled = pyqtSignal() + error = pyqtSignal(str) + + def __init__(self, preview_service, repo) -> None: + super().__init__() + self._preview_service = preview_service + self._repo = repo + self._cancel = threading.Event() + + @pyqtSlot() + def cancel(self) -> None: + self._cancel.set() + + @pyqtSlot(AutocropBatchTask) + def process(self, task: AutocropBatchTask) -> None: + from negpy.desktop.session import resolve_asset_rgbscan + from negpy.features.flatfield.logic import apply_flatfield + from negpy.features.geometry.batch_autocrop import ( + BatchDetectParams, + detect_autocrop_candidate, + finalize_autocrop_batch, + prepare_autocrop_image, + ) + + self._cancel.clear() + candidates = [] + configs: dict[str, WorkspaceConfig] = {} + skipped: set[str] = set() + failed: dict[str, str] = {} + total = len(task.files) + params = BatchDetectParams(frame_ratio=task.frame_ratio) + + try: + for index, file_info in enumerate(task.files): + if self._cancel.is_set(): + self.cancelled.emit() + return + + source_hash = file_info["hash"] + config = resolve_asset_rgbscan( + self._repo.load_file_settings(source_hash) or task.default_config, + file_info, + ) + geometry = config.geometry + if geometry.manual_crop_rect is not None: + skipped.add(source_hash) + self.progress.emit(index + 1, total, f"{file_info['name']} [manual crop]") + continue + + try: + rgbscan = config.rgbscan + if rgbscan.enabled and rgbscan.green_path and rgbscan.blue_path: + image, _, _ = self._preview_service.load_linear_preview_rgb( + file_info["path"], + rgbscan.green_path, + rgbscan.blue_path, + task.workspace_color_space, + use_camera_wb=not config.process.linear_raw, + full_resolution=False, + file_hash=source_hash, + align=rgbscan.align, + ) + else: + image, _, _ = self._preview_service.load_linear_preview( + file_info["path"], + task.workspace_color_space, + not config.process.linear_raw, + False, + source_hash, + ) + image = apply_flatfield(image, config.flatfield) + transformed = prepare_autocrop_image( + image, + geometry, + config.flatfield.k1 if config.flatfield.apply else 0.0, + ) + candidates.append(detect_autocrop_candidate(transformed, source_hash, params)) + configs[source_hash] = config + self.progress.emit(index + 1, total, file_info["name"]) + except Exception as exc: + logger.error("Auto Crop All failed for %s: %s", file_info["name"], exc) + failed[source_hash] = str(exc) + self.progress.emit(index + 1, total, f"{file_info['name']} [failed]") + + if self._cancel.is_set(): + self.cancelled.emit() + return + if not candidates: + raise RuntimeError("No uncropped files were available for Auto Crop All") + + self.finished.emit( + { + "results": finalize_autocrop_batch(candidates, params), + "canvases": {candidate.source_id: candidate.canvas for candidate in candidates}, + "configs": configs, + "skipped": skipped, + "failed": failed, + } + ) + except Exception as exc: + logger.error("Auto Crop All failure: %s", exc) + self.error.emit(str(exc)) diff --git a/negpy/features/geometry/batch_autocrop.py b/negpy/features/geometry/batch_autocrop.py new file mode 100644 index 00000000..5955d486 --- /dev/null +++ b/negpy/features/geometry/batch_autocrop.py @@ -0,0 +1,564 @@ +"""Rotation-aware film-frame detection and batch calibration.""" + +from __future__ import annotations + +from dataclasses import dataclass, replace + +import cv2 +import numpy as np + +from negpy.domain.models import WorkspaceConfig +from negpy.domain.types import ImageBuffer, ROI +from negpy.features.geometry.models import FINE_ROTATION_LIMIT, GeometryConfig + +ANALYZE_MAX = 1200 +ANGLE_ANALYZE_MAX = 500 + + +@dataclass(frozen=True) +class BatchDetectParams: + max_frac: float = 0.35 + dark_thresh: float = 0.12 + bright_thresh: float = 0.985 + start_frac: float = 0.60 + strong_frac: float = 0.82 + gap: int = 8 + inset: int = 4 + max_angle: float = 5.0 + min_mask_sides: int = 2 + frame_ratio: float = 1.5 + outward_frac: float = 0.01 + + +@dataclass(frozen=True) +class AutocropCandidate: + source_id: str + roi: ROI + canvas: tuple[int, int] + angle: float + plausible_geometry: bool + top_found: bool + left_found: bool + right_found: bool + left_primary: bool = False + right_primary: bool = False + angle_confident: bool = True + vertical_profile: np.ndarray | None = None + + +@dataclass(frozen=True) +class AutocropResult: + roi: ROI + angle: float + method: str + + +@dataclass(frozen=True) +class BatchTemplate: + width: int + fallback_width: int + width_tolerance: int + center_x: int + top: int + angle: float + angle_tolerance: float + + +def autocrop_input_signature(config: WorkspaceConfig) -> tuple: + """Settings that must remain stable between detection and persistence.""" + geometry = config.geometry + return ( + geometry.rotation, + geometry.fine_rotation, + geometry.flip_horizontal, + geometry.flip_vertical, + config.process.linear_raw, + config.rgbscan, + config.flatfield, + ) + + +def geometry_from_autocrop_result( + geometry: GeometryConfig, + result: AutocropResult, + canvas: tuple[int, int], +) -> GeometryConfig | None: + """Convert a calibrated result into persistent NegPy geometry settings.""" + if result.method == "manual": + return None + fine_rotation = geometry.fine_rotation + result.angle + if abs(fine_rotation) > FINE_ROTATION_LIMIT: + return None + height, width = canvas + top, bottom, left, right = result.roi + manual_rect = (left / width, top / height, right / width, bottom / height) + return replace( + geometry, + fine_rotation=fine_rotation, + manual_crop_rect=manual_rect, + auto_crop_enabled=False, + ) + + +def _validate_params(params: BatchDetectParams) -> None: + if not 0 < params.max_frac < 0.5: + raise ValueError("max_frac must be between 0 and 0.5") + if params.inset < 0: + raise ValueError("inset must be non-negative") + if not 0 <= params.max_angle <= 15: + raise ValueError("max_angle must be between 0 and 15") + if params.frame_ratio <= 0: + raise ValueError("frame_ratio must be positive") + if not 0 <= params.outward_frac <= 0.1: + raise ValueError("outward_frac must be between 0 and 0.1") + + +def _analysis_luminance(image: ImageBuffer) -> tuple[np.ndarray, tuple[int, int], tuple[float, float]]: + height, width = image.shape[:2] + scale = min(1.0, ANALYZE_MAX / max(width, height)) + analysis = image + if scale < 1: + analysis = cv2.resize( + np.ascontiguousarray(image), + None, + fx=scale, + fy=scale, + interpolation=cv2.INTER_AREA, + ) + if analysis.ndim == 2: + gray = analysis + elif analysis.shape[2] == 4: + gray = cv2.cvtColor(analysis, cv2.COLOR_RGBA2GRAY) + else: + gray = cv2.cvtColor(analysis, cv2.COLOR_RGB2GRAY) + white = float(np.percentile(gray, 99)) + if white <= 0: + normalized = np.zeros(gray.shape, dtype=np.float32) + else: + normalized = np.clip(gray.astype(np.float32) * (255.0 / white), 0, 255) + return normalized, (height, width), (normalized.shape[0] / height, normalized.shape[1] / width) + + +def _rotate(image: np.ndarray, angle: float) -> np.ndarray: + if abs(angle) < 0.01: + return image.copy() + height, width = image.shape[:2] + matrix = cv2.getRotationMatrix2D((width / 2, height / 2), angle, 1.0) + return cv2.warpAffine( + image, + matrix, + (width, height), + flags=cv2.INTER_LINEAR, + borderMode=cv2.BORDER_REPLICATE, + ) + + +def _robust_profile(gradient: np.ndarray, axis: int) -> np.ndarray: + if axis == 1: + gradient = gradient[:, int(gradient.shape[1] * 0.08) : int(gradient.shape[1] * 0.92)] + else: + gradient = gradient[int(gradient.shape[0] * 0.08) : int(gradient.shape[0] * 0.92)] + cap = np.quantile(gradient, 0.85) + return np.quantile(gradient, 0.55, axis=axis) + 0.15 * np.mean(np.minimum(gradient, cap), axis=axis) + + +def _edge_profiles(gray: np.ndarray, angle: float) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]: + rotated = _rotate(gray, angle) + blurred = cv2.GaussianBlur(rotated, (0, 0), 2.0) + gradient_x = np.abs(cv2.Sobel(blurred, cv2.CV_32F, 1, 0, ksize=3)) + gradient_y = np.abs(cv2.Sobel(blurred, cv2.CV_32F, 0, 1, ksize=3)) + return ( + rotated, + _robust_profile(gradient_y, 1), + _robust_profile(gradient_x, 0), + gradient_x, + gradient_y, + ) + + +def _angle_score(gray: np.ndarray, angle: float, max_frac: float) -> float: + _, horizontal, vertical, _, _ = _edge_profiles(gray, angle) + height, width = gray.shape + bands = ( + horizontal[3 : int(height * max_frac)], + horizontal[int(height * (1 - max_frac)) : -3], + vertical[3 : int(width * max_frac)], + vertical[int(width * (1 - max_frac)) : -3], + ) + scores = [float(np.max(band)) for band in bands if band.size] + return sum(scores) + min(scores) + + +def _estimate_angle(gray: np.ndarray, params: BatchDetectParams) -> float: + scale = min(1.0, ANGLE_ANALYZE_MAX / max(gray.shape)) + small = cv2.resize(gray, None, fx=scale, fy=scale, interpolation=cv2.INTER_AREA) if scale < 1 else gray + coarse = np.arange(-params.max_angle, params.max_angle + 0.01, 0.5) + coarse_scores = np.asarray([_angle_score(small, angle, params.max_frac) for angle in coarse]) + center = float(np.mean(coarse[coarse_scores >= coarse_scores.max() * 0.999])) + fine = np.arange( + max(-params.max_angle, center - 0.5), + min(params.max_angle, center + 0.5) + 0.01, + 0.1, + ) + fine_scores = np.asarray([_angle_score(small, angle, params.max_frac) for angle in fine]) + return round(float(np.mean(fine[fine_scores >= fine_scores.max() * 0.999])), 2) + + +def _edge_line_fit(gray: np.ndarray, angle: float, left: int, right: int, edge: int) -> tuple[float, float, int] | None: + rotated = _rotate(gray, angle) + blurred = cv2.GaussianBlur(rotated, (0, 0), 1.5) + gradient = np.abs(cv2.Sobel(blurred, cv2.CV_32F, 0, 1, ksize=3)) + radius = max(12, round(gray.shape[0] * 0.035)) + low = max(2, edge - radius) + high = min(gray.shape[0] - 2, edge + radius) + xs = np.arange(left + 20, right - 20, 3) + if len(xs) < 20 or high <= low: + return None + ys = np.asarray([low + np.argmax(gradient[low:high, x]) for x in xs]) + scores = gradient[ys, xs] + points = np.column_stack([xs, ys])[scores >= np.quantile(scores, 0.60)] + if len(points) < 20: + return None + residuals = np.zeros(len(points)) + slope = 0.0 + for _ in range(5): + slope, intercept = np.polyfit(points[:, 0], points[:, 1], 1) + residuals = np.abs(points[:, 1] - (slope * points[:, 0] + intercept)) + keep = residuals <= max(2.0, float(np.median(residuals)) * 2.5) + if keep.all() or keep.sum() < 20: + break + points = points[keep] + median_residual = float(np.median(residuals)) + if median_residual > 2.0: + return None + return float(np.degrees(np.arctan(slope))), median_residual, len(points) + + +def _refine_angle( + gray: np.ndarray, + angle: float, + edges: tuple[int, int, int, int], + params: BatchDetectParams, +) -> tuple[float, bool]: + left, top, right, _ = edges + if right - left < gray.shape[1] * 0.4: + return angle, False + top_fit = _edge_line_fit(gray, angle, left, right, top) + min_points = max(20, round((right - left) * 0.08)) + if top_fit is None or abs(top_fit[0]) > 0.5 or top_fit[2] < min_points: + return angle, False + return round(min(max(angle + top_fit[0], -params.max_angle), params.max_angle), 2), True + + +def _mask_depth(fractions: np.ndarray, limit: int, params: BatchDetectParams) -> int | None: + if limit <= 0 or fractions[: min(8, limit)].max() < params.start_frac: + return None + depth = 0 + for index in range(limit): + if fractions[index] >= params.strong_frac: + depth = index + 1 + elif index - depth > params.gap: + break + if depth == 0 or depth >= limit - 2: + return None + return depth + + +def _strongest_edge(profile: np.ndarray, start: int, end: int) -> int: + return int(np.argmax(profile[start:end])) + start + + +def _fallback_is_border(gray: np.ndarray, gradient: np.ndarray, edge: int, side: str) -> bool: + height, width = gray.shape + margin = 3 + if side == "top": + outside = gray[margin : max(margin + 1, edge - 2)] + texture = gradient[margin : max(margin + 1, edge - 2)] + span = edge + elif side == "bottom": + outside = gray[min(height - 1, edge + 2) : height - margin] + texture = gradient[min(height - 1, edge + 2) : height - margin] + span = height - edge + elif side == "left": + outside = gray[:, margin : max(margin + 1, edge - 2)] + texture = gradient[:, margin : max(margin + 1, edge - 2)] + span = edge + else: + outside = gray[:, min(width - 1, edge + 2) : width - margin] + texture = gradient[:, min(width - 1, edge + 2) : width - margin] + span = width - edge + if outside.size == 0 or span < min(height, width) * 0.008: + return False + median = float(np.median(outside)) + return (median < 56 or median > 245) and float(np.median(texture)) < 12 + + +def _is_coherent_outer_edge(profile: np.ndarray, edge: int, side: str) -> bool: + axis_length = len(profile) + span = edge if side in {"top", "left"} else axis_length - edge + if span > axis_length * 0.03: + return False + outer_limit = max(4, round(axis_length * 0.35)) + band = profile[3:outer_limit] if side in {"top", "left"} else profile[axis_length - outer_limit : axis_length - 3] + return bool(band.size and float(profile[edge]) >= max(20.0, float(np.quantile(band, 0.95)) * 2.5)) + + +def _analysis_edges( + gray: np.ndarray, + angle: float, + params: BatchDetectParams, +) -> tuple[tuple[int, int, int, int], frozenset[str]] | None: + rotated, horizontal, vertical, gradient_x, gradient_y = _edge_profiles(gray, angle) + height, width = rotated.shape + border = (rotated < params.dark_thresh * 255) | (rotated > params.bright_thresh * 255) + limit_y = int(height * params.max_frac) + limit_x = int(width * params.max_frac) + mask_depths = { + "top": _mask_depth(border.mean(axis=1), limit_y, params), + "bottom": _mask_depth(border[::-1].mean(axis=1), limit_y, params), + "left": _mask_depth(border.mean(axis=0), limit_x, params), + "right": _mask_depth(border[:, ::-1].mean(axis=0), limit_x, params), + } + if sum(depth is not None for depth in mask_depths.values()) < params.min_mask_sides: + return None + fallback_edges = { + "top": _strongest_edge(horizontal, 3, limit_y), + "bottom": _strongest_edge(horizontal, height - limit_y, height - 3), + "left": _strongest_edge(vertical, 3, limit_x), + "right": _strongest_edge(vertical, width - limit_x, width - 3), + } + edges: dict[str, int] = {} + for side in ("top", "bottom", "left", "right"): + depth = mask_depths[side] + axis_length = height if side in {"top", "bottom"} else width + if depth is not None: + mask_edge = depth if side in {"top", "left"} else axis_length - depth + gradient_edge = fallback_edges[side] + inward_delta = gradient_edge - mask_edge if side in {"top", "left"} else mask_edge - gradient_edge + edges[side] = gradient_edge if 0 < inward_delta <= axis_length * 0.03 else mask_edge + continue + candidate = fallback_edges[side] + gradient = gradient_y if side in {"top", "bottom"} else gradient_x + profile = horizontal if side in {"top", "bottom"} else vertical + if _fallback_is_border(rotated, gradient, candidate, side) or _is_coherent_outer_edge(profile, candidate, side): + edges[side] = candidate + else: + edges[side] = 0 if side in {"top", "left"} else axis_length + inset = params.inset + return ( + ( + min(edges["left"] + (inset if edges["left"] else 0), width), + min(edges["top"] + (inset if edges["top"] else 0), height), + max(edges["right"] - (inset if edges["right"] < width else 0), 0), + max(edges["bottom"] - (inset if edges["bottom"] < height else 0), 0), + ), + frozenset(side for side, depth in mask_depths.items() if depth is not None), + ) + + +def detect_autocrop_candidate( + image: ImageBuffer, + source_id: str, + params: BatchDetectParams | None = None, +) -> AutocropCandidate: + selected = params or BatchDetectParams() + _validate_params(selected) + gray, (height, width), (scale_y, scale_x) = _analysis_luminance(image) + angle = _estimate_angle(gray, selected) + edge_detection = _analysis_edges(gray, angle, selected) + if edge_detection is None: + return AutocropCandidate(source_id, (0, height, 0, width), (height, width), angle, False, False, False, False) + edges, primary_sides = edge_detection + left, top, right, bottom = edges + edge_ratio = (right - left) / max(1, bottom - top) + plausible_geometry = selected.frame_ratio * 0.8 <= edge_ratio <= selected.frame_ratio * 1.2 + angle_confident = False + if plausible_geometry: + angle, angle_confident = _refine_angle(gray, angle, edges, selected) + edge_detection = _analysis_edges(gray, angle, selected) + if edge_detection is None: + return AutocropCandidate(source_id, (0, height, 0, width), (height, width), angle, False, False, False, False) + edges, primary_sides = edge_detection + left, top, right, bottom = edges + roi = ( + round(top / scale_y), + round(bottom / scale_y), + round(left / scale_x), + round(right / scale_x), + ) + if roi[1] <= roi[0] or roi[3] <= roi[2]: + raise ValueError("detected crop has no usable area") + return AutocropCandidate( + source_id=source_id, + roi=roi, + canvas=(height, width), + angle=angle, + plausible_geometry=plausible_geometry, + top_found=top > 0, + left_found=left > 0, + right_found=right < gray.shape[1], + left_primary="left" in primary_sides, + right_primary="right" in primary_sides, + angle_confident=angle_confident, + vertical_profile=_edge_profiles(gray, angle)[2], + ) + + +def _bounded_interval(center: float, length: int, limit: int) -> tuple[int, int]: + length = min(length, limit) + start = min(max(0, round(center - length / 2)), limit - length) + return start, start + length + + +def prepare_autocrop_image( + image: ImageBuffer, + geometry: GeometryConfig, + distortion_k1: float = 0.0, +) -> ImageBuffer: + """Apply the geometry that precedes crop detection in the render pipeline.""" + from negpy.features.geometry.logic import apply_fine_rotation, apply_radial_distortion + + prepared = image + if geometry.rotation: + prepared = np.rot90(prepared, k=geometry.rotation) + if geometry.flip_horizontal: + prepared = np.fliplr(prepared) + if geometry.flip_vertical: + prepared = np.flipud(prepared) + prepared = np.ascontiguousarray(prepared) + if geometry.fine_rotation: + prepared = apply_fine_rotation(prepared, geometry.fine_rotation) + if distortion_k1: + prepared = apply_radial_distortion(prepared, distortion_k1) + return prepared + + +def _build_templates(candidates: list[AutocropCandidate]) -> dict[bool, BatchTemplate]: + templates: dict[bool, BatchTemplate] = {} + for landscape in (False, True): + trusted = [ + candidate + for candidate in candidates + if (candidate.canvas[1] >= candidate.canvas[0]) == landscape + and candidate.plausible_geometry + and candidate.top_found + and candidate.left_found + and candidate.right_found + ] + if not trusted: + continue + all_widths = np.asarray([item.roi[3] - item.roi[2] for item in trusted]) + median_width = float(np.median(all_widths)) + width_mad = float(np.median(np.abs(all_widths - median_width))) + width_tolerance = max(median_width * 0.01, width_mad * 3) + trusted = [item for item in trusted if abs((item.roi[3] - item.roi[2]) - median_width) <= width_tolerance] + widths = np.asarray([item.roi[3] - item.roi[2] for item in trusted]) + angles = np.asarray([item.angle for item in trusted]) + median_angle = float(np.median(angles)) + angle_mad = float(np.median(np.abs(angles - median_angle))) + angle_tolerance = max(0.35, angle_mad * 3) + angle_inliers = angles[np.abs(angles - median_angle) <= angle_tolerance] + templates[landscape] = BatchTemplate( + width=round(float(np.median(widths))), + fallback_width=round(float(np.quantile(widths, 0.9))), + width_tolerance=round(width_tolerance), + center_x=round(float(np.median([(item.roi[2] + item.roi[3]) / 2 for item in trusted]))), + top=round(float(np.median([item.roi[0] for item in trusted]))), + angle=round(float(np.median(angle_inliers)), 2), + angle_tolerance=angle_tolerance, + ) + return templates + + +def _finalize_candidate( + candidate: AutocropCandidate, + template: BatchTemplate | None, + frame_ratio: float, +) -> AutocropResult: + height, width = candidate.canvas + top, _, left, right = candidate.roi + full = (0, height, 0, width) + has_two_corners = candidate.plausible_geometry and candidate.top_found and candidate.left_found and candidate.right_found + measured_width = right - left + if has_two_corners and template is not None: + has_two_corners = abs(measured_width - template.width) <= template.width_tolerance + if has_two_corners: + adjusted_width = max(measured_width, template.width if template else measured_width) + if template and measured_width < template.width - template.width_tolerance / 2: + adjusted_width = template.fallback_width + adjusted_angle = ( + template.angle + if template + and (not candidate.angle_confident or abs(candidate.angle - template.angle) > template.angle_tolerance) + else candidate.angle + ) + bottom = top + round(adjusted_width / frame_ratio) + if left < 0 or top < 0 or left + adjusted_width > width or bottom > height: + return AutocropResult(full, 0.0, "manual") + return AutocropResult((top, bottom, left, left + adjusted_width), adjusted_angle, "two-corner") + if template is None or candidate.roi == full: + return AutocropResult(full, 0.0, "manual") + core_width = min(width, template.fallback_width) + core_height = min(height, round(core_width / frame_ratio)) + corner_count = int(candidate.left_found) + int(candidate.right_found) if candidate.top_found else 0 + anchor_left = candidate.left_found + anchor_right = candidate.right_found + if corner_count == 2 and candidate.plausible_geometry and candidate.left_primary != candidate.right_primary: + corner_count = 1 + anchor_left = candidate.left_primary + anchor_right = candidate.right_primary + if corner_count == 1 and anchor_left: + left, right = left, left + core_width + elif corner_count == 1 and anchor_right: + left, right = right - core_width, right + else: + left, right = _bounded_interval(template.center_x, core_width, width) + if left < 0 or right > width: + left, right = _bounded_interval(template.center_x, core_width, width) + corner_count = 0 + use_detected_top = corner_count == 1 and abs(top - template.top) <= height * 0.05 + top = top if use_detected_top else template.top + angle = candidate.angle if corner_count == 1 and candidate.angle_confident else template.angle + bottom = top + core_height + if top < 0 or bottom > height: + return AutocropResult(full, 0.0, "manual") + return AutocropResult((top, bottom, round(left), round(right)), angle, "one-corner" if corner_count == 1 else "batch-template") + + +def _add_outward_margin( + result: AutocropResult, + canvas: tuple[int, int], + outward_frac: float, +) -> AutocropResult: + if result.method == "manual" or outward_frac == 0: + return result + height, width = canvas + top, bottom, left, right = result.roi + requested = round(min(height, width) * outward_frac) + margin = max(0, min(requested, top, left, height - bottom, width - right)) + return AutocropResult( + (top - margin, bottom + margin, left - margin, right + margin), + result.angle, + result.method, + ) + + +def finalize_autocrop_batch( + candidates: list[AutocropCandidate], + params: BatchDetectParams | None = None, +) -> dict[str, AutocropResult]: + selected = params or BatchDetectParams() + _validate_params(selected) + templates = _build_templates(candidates) + return { + candidate.source_id: _add_outward_margin( + _finalize_candidate( + candidate, + templates.get(candidate.canvas[1] >= candidate.canvas[0]), + selected.frame_ratio, + ), + candidate.canvas, + selected.outward_frac, + ) + for candidate in candidates + } diff --git a/tests/test_batch_autocrop.py b/tests/test_batch_autocrop.py new file mode 100644 index 00000000..5dc07298 --- /dev/null +++ b/tests/test_batch_autocrop.py @@ -0,0 +1,163 @@ +import cv2 +import numpy as np + +from negpy.features.geometry.batch_autocrop import ( + AutocropCandidate, + AutocropResult, + BatchDetectParams, + detect_autocrop_candidate, + finalize_autocrop_batch, + geometry_from_autocrop_result, +) +from negpy.features.geometry.models import GeometryConfig + + +def _candidate( + source_id: str, + roi: tuple[int, int, int, int], + *, + angle: float = 0.3, + plausible: bool = True, + top: bool = True, + left: bool = True, + right: bool = True, + left_primary: bool = False, + right_primary: bool = False, +) -> AutocropCandidate: + return AutocropCandidate( + source_id=source_id, + roi=roi, + canvas=(200, 300), + angle=angle, + plausible_geometry=plausible, + top_found=top, + left_found=left, + right_found=right, + left_primary=left_primary, + right_primary=right_primary, + ) + + +def test_detect_candidate_accepts_in_memory_float_image() -> None: + image = np.full((400, 600, 3), 0.5, dtype=np.float32) + image[:30] = 0.0 + image[-25:] = 1.0 + image[:, :40] = 0.0 + image[:, -45:] = 0.0 + + candidate = detect_autocrop_candidate(image, "scan", BatchDetectParams(frame_ratio=1.5)) + + assert candidate.source_id == "scan" + assert candidate.canvas == (400, 600) + assert candidate.plausible_geometry + assert candidate.top_found + assert candidate.left_found + assert candidate.right_found + + +def test_detect_candidate_estimates_rotation() -> None: + image = np.full((400, 600, 3), 0.5, dtype=np.float32) + image[:30] = 0.0 + image[-25:] = 1.0 + image[:, :40] = 0.0 + image[:, -45:] = 0.0 + matrix = cv2.getRotationMatrix2D((300, 200), -2.0, 1.0) + tilted = cv2.warpAffine(image, matrix, (600, 400), borderMode=cv2.BORDER_REPLICATE) + + candidate = detect_autocrop_candidate(tilted, "scan") + + assert abs(candidate.angle - 2.0) <= 0.15 + + +def test_batch_preserves_one_corner_and_unsupported_image() -> None: + reference = _candidate("reference", (14, 184, 22, 279)) + one_corner = _candidate("one-corner", (14, 184, 24, 300), right=False) + unsupported = _candidate( + "unsupported", + (0, 200, 0, 300), + angle=0.0, + plausible=False, + top=False, + left=False, + right=False, + ) + + results = finalize_autocrop_batch([reference, one_corner, unsupported]) + + assert results["reference"].method == "two-corner" + assert results["one-corner"].method == "one-corner" + assert results["one-corner"].roi[2] == 22 + assert results["unsupported"].method == "manual" + assert results["unsupported"].roi == (0, 200, 0, 300) + + +def test_batch_rejects_width_and_angle_outliers() -> None: + references = [ + _candidate(f"reference-{index}", (14, 184, 20, 20 + width), angle=angle) + for index, (width, angle) in enumerate(((255, 0.2), (257, 0.3), (259, 0.4))) + ] + outlier = _candidate("outlier", (14, 184, 4, 296), angle=-0.5) + + result = finalize_autocrop_batch([*references, outlier])["outlier"] + + assert result.method == "batch-template" + assert result.roi[3] - result.roi[2] < 292 + assert result.angle == 0.3 + + +def test_batch_uses_only_primary_corner_for_false_edge() -> None: + references = [ + _candidate( + f"reference-{index}", + (14, 184, 20, 20 + width), + left_primary=True, + right_primary=True, + ) + for index, width in enumerate((255, 257, 259)) + ] + false_right = _candidate( + "false-right", + (14, 184, 12, 296), + left_primary=True, + right_primary=False, + ) + + result = finalize_autocrop_batch([*references, false_right])["false-right"] + + assert result.method == "one-corner" + assert result.roi[2] == 10 + + +def test_batch_keeps_equal_outward_safety_margin() -> None: + candidate = _candidate("reference", (14, 184, 22, 279)) + + exact = finalize_autocrop_batch( + [candidate], + BatchDetectParams(outward_frac=0), + )["reference"] + padded = finalize_autocrop_batch([candidate])["reference"] + + assert padded.roi == ( + exact.roi[0] - 2, + exact.roi[1] + 2, + exact.roi[2] - 2, + exact.roi[3] + 2, + ) + + +def test_result_becomes_persistent_manual_crop_and_rotation() -> None: + geometry = GeometryConfig(fine_rotation=0.5, auto_crop_enabled=True) + result = AutocropResult((20, 180, 30, 270), 1.25, "two-corner") + + updated = geometry_from_autocrop_result(geometry, result, (200, 300)) + + assert updated is not None + assert updated.fine_rotation == 1.75 + assert updated.manual_crop_rect == (0.1, 0.1, 0.9, 0.9) + assert not updated.auto_crop_enabled + + +def test_manual_result_leaves_geometry_unchanged() -> None: + result = AutocropResult((0, 200, 0, 300), 0.0, "manual") + + assert geometry_from_autocrop_result(GeometryConfig(), result, (200, 300)) is None diff --git a/tests/test_batch_progress.py b/tests/test_batch_progress.py index d691b471..1bfcf610 100644 --- a/tests/test_batch_progress.py +++ b/tests/test_batch_progress.py @@ -1,4 +1,7 @@ -from unittest.mock import MagicMock +from dataclasses import replace +from unittest.mock import MagicMock, patch + +import numpy as np from PyQt6.QtCore import Qt from PyQt6.QtTest import QTest @@ -6,7 +9,13 @@ from negpy.desktop.view.widgets.progress_dialog import ProgressDialog from negpy.desktop.workers.export import ExportTask, ExportWorker -from negpy.desktop.workers.render import NormalizationTask, NormalizationWorker +from negpy.desktop.workers.render import ( + AutocropBatchTask, + AutocropBatchWorker, + NormalizationTask, + NormalizationWorker, +) +from negpy.features.geometry.batch_autocrop import AutocropCandidate from negpy.kernel.system.config import DEFAULT_WORKSPACE_CONFIG @@ -112,6 +121,112 @@ def _task(name: str) -> ExportTask: assert not _same_decode_source(a, a_flat) +def test_autocrop_all_skips_manual_crops_and_returns_calibrated_results() -> None: + preview = MagicMock() + preview.load_linear_preview.return_value = ( + np.zeros((200, 300, 3), dtype=np.float32), + (200, 300), + {}, + ) + repo = MagicMock() + manual = replace( + DEFAULT_WORKSPACE_CONFIG, + geometry=replace( + DEFAULT_WORKSPACE_CONFIG.geometry, + manual_crop_rect=(0.1, 0.1, 0.9, 0.9), + ), + ) + repo.load_file_settings.side_effect = lambda source_hash: manual if source_hash == "manual" else None + worker = AutocropBatchWorker(preview, repo) + task = AutocropBatchTask( + files=[ + {"name": "manual.tif", "path": "/tmp/manual.tif", "hash": "manual"}, + {"name": "auto.tif", "path": "/tmp/auto.tif", "hash": "auto"}, + ], + workspace_color_space="Adobe RGB", + default_config=DEFAULT_WORKSPACE_CONFIG, + frame_ratio=1.5, + ) + candidate = AutocropCandidate( + source_id="auto", + roi=(14, 184, 22, 277), + canvas=(200, 300), + angle=0.3, + plausible_geometry=True, + top_found=True, + left_found=True, + right_found=True, + ) + completed: list[dict] = [] + worker.finished.connect(completed.append) + + with patch( + "negpy.features.geometry.batch_autocrop.detect_autocrop_candidate", + return_value=candidate, + ): + worker.process(task) + + assert len(completed) == 1 + assert completed[0]["skipped"] == {"manual"} + assert completed[0]["results"]["auto"].method == "two-corner" + preview.load_linear_preview.assert_called_once() + + +def test_autocrop_all_uses_each_assets_rgb_triplet() -> None: + from negpy.features.rgbscan.models import RgbScanConfig + + preview = MagicMock() + preview.load_linear_preview_rgb.return_value = ( + np.zeros((200, 300, 3), dtype=np.float32), + (200, 300), + {}, + ) + repo = MagicMock() + repo.load_file_settings.return_value = None + worker = AutocropBatchWorker(preview, repo) + default_config = replace( + DEFAULT_WORKSPACE_CONFIG, + rgbscan=RgbScanConfig( + enabled=True, + green_path="/current-green.tif", + blue_path="/current-blue.tif", + ), + ) + task = AutocropBatchTask( + files=[ + { + "name": "frame.tif", + "path": "/frame-red.tif", + "hash": "frame", + "green_path": "/frame-green.tif", + "blue_path": "/frame-blue.tif", + } + ], + workspace_color_space="Adobe RGB", + default_config=default_config, + frame_ratio=1.5, + ) + candidate = AutocropCandidate( + source_id="frame", + roi=(14, 184, 22, 277), + canvas=(200, 300), + angle=0.0, + plausible_geometry=True, + top_found=True, + left_found=True, + right_found=True, + ) + + with patch( + "negpy.features.geometry.batch_autocrop.detect_autocrop_candidate", + return_value=candidate, + ): + worker.process(task) + + args = preview.load_linear_preview_rgb.call_args.args + assert args[:3] == ("/frame-red.tif", "/frame-green.tif", "/frame-blue.tif") + + def test_normalization_worker_cancel_emits_cancelled_no_baseline() -> None: preview = MagicMock() repo = MagicMock() diff --git a/tests/test_controller.py b/tests/test_controller.py index 99010283..157aa5de 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -445,6 +445,7 @@ def setUp(self): self.mock_session_manager.state = AppState() self.mock_session_manager.repo = MagicMock() self.mock_session_manager.repo.load_file_settings.return_value = None + self.mock_session_manager.fresh_file_config.return_value = WorkspaceConfig() self.mock_session_manager.state.uploaded_files = [ {"name": "IMG_0001.cr2", "path": "/tmp/IMG_0001.cr2", "hash": "h1"}, @@ -948,6 +949,7 @@ def setUp(self): self.mock_session_manager.state = AppState() self.mock_session_manager.repo = MagicMock() self.mock_session_manager.repo.load_file_settings.return_value = None + self.mock_session_manager.fresh_file_config.return_value = WorkspaceConfig() self.mock_session_manager.state.uploaded_files = [ {"name": "IMG_0001.cr2", "path": "/tmp/IMG_0001.cr2", "hash": "h1"}, @@ -970,6 +972,9 @@ def setUp(self): self.emitted = [] self.controller.normalization_requested.connect(self.emitted.append) + self.autocrop_emitted = [] + self.controller.autocrop_batch_requested.disconnect(self.controller.autocrop_batch_worker.process) + self.controller.autocrop_batch_requested.connect(self.autocrop_emitted.append) def tearDown(self): import gc @@ -1006,6 +1011,90 @@ def test_analysis_zero_matches_does_not_dispatch(self): self.controller.request_batch_normalization() self.assertEqual(self.emitted, []) + def test_analysis_does_not_start_during_auto_crop_all(self): + self.controller._active_batch = "Auto cropping roll" + + self.controller.request_batch_normalization() + + self.assertEqual(self.emitted, []) + + def test_auto_crop_all_respects_filter(self): + self.visible_indices = [1, 2] + with patch("negpy.desktop.controller.QMessageBox") as mock_box: + mock_box.StandardButton.Yes = 1 + mock_box.question.return_value = 1 + self.controller.apply_auto_crop_all() + + self.assertEqual(len(self.autocrop_emitted), 1) + self.assertEqual( + [file_info["name"] for file_info in self.autocrop_emitted[0].files], + ["IMG_0002.cr2", "scan.tif"], + ) + self.assertEqual(self.autocrop_emitted[0].default_config, WorkspaceConfig()) + + def test_auto_crop_all_persists_explicit_crop(self): + from negpy.features.geometry.batch_autocrop import AutocropResult + + self.mock_session_manager.state.current_file_hash = "h1" + config = self.mock_session_manager.state.config + payload = { + "results": {"h1": AutocropResult((20, 180, 30, 270), 1.25, "two-corner")}, + "canvases": {"h1": (200, 300)}, + "configs": {"h1": config}, + "skipped": set(), + "failed": {}, + } + + self.controller._on_autocrop_batch_finished(payload) + + saved = self.mock_session_manager.repo.save_file_settings.call_args.args[1] + self.assertEqual(saved.geometry.manual_crop_rect, (0.1, 0.1, 0.9, 0.9)) + self.assertEqual(saved.geometry.fine_rotation, 1.25) + self.assertFalse(saved.geometry.auto_crop_enabled) + self.mock_session_manager.update_config.assert_called_once() + + def test_auto_crop_all_preserves_manual_crop_added_while_running(self): + from negpy.features.geometry.batch_autocrop import AutocropResult + + captured = self.mock_session_manager.state.config + latest = replace( + captured, + geometry=replace(captured.geometry, manual_crop_rect=(0.2, 0.2, 0.8, 0.8)), + ) + self.mock_session_manager.repo.load_file_settings.return_value = latest + payload = { + "results": {"h1": AutocropResult((20, 180, 30, 270), 1.25, "two-corner")}, + "canvases": {"h1": (200, 300)}, + "configs": {"h1": captured}, + "skipped": set(), + "failed": {}, + } + + self.controller._on_autocrop_batch_finished(payload) + + self.mock_session_manager.repo.save_file_settings.assert_not_called() + + def test_auto_crop_all_skips_geometry_changed_while_running(self): + from negpy.features.geometry.batch_autocrop import AutocropResult + + captured = self.mock_session_manager.state.config + latest = replace( + captured, + geometry=replace(captured.geometry, fine_rotation=2.0), + ) + self.mock_session_manager.repo.load_file_settings.return_value = latest + payload = { + "results": {"h1": AutocropResult((20, 180, 30, 270), 1.25, "two-corner")}, + "canvases": {"h1": (200, 300)}, + "configs": {"h1": captured}, + "skipped": set(), + "failed": {}, + } + + self.controller._on_autocrop_batch_finished(payload) + + self.mock_session_manager.repo.save_file_settings.assert_not_called() + class TestContactSheetOutputDir(unittest.TestCase): def setUp(self):