diff --git a/VERSION b/VERSION index 9b0025a7..72a8a631 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.40.0 +0.41.0 diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 05ddd45d..4e8d9a66 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -1,5 +1,9 @@ # Change Log +## 0.41.0 + +- New: **Stitch multi-shot scans** — select overlapping shots of one frame (e.g. a 6×6 scanned in two halves) on the contact sheet and pick **Stitch selected frames**. Alignment, exposure matching and blending happen on the linear scan data before conversion, so the result develops like a single raw. No new file is written: the composite edits and exports like any frame, and **Unstitch** restores the parts. IR dust data is kept when all parts have it. + ## 0.40.0 - New: **Sharpening rebuilt — Radius, Masking and a new Deconvolution mode** — the Sharpen controls, now in their own section, gain a **Radius** slider (how fine or broad the sharpened detail is), a **Masking** slider (holds sharpening off flat areas like skies and skin so grain and noise stay quiet), and a **Method** dropdown: **Unsharp Mask** (the classic, now with halo suppression on high-contrast edges) or **Deconvolution** (Richardson-Lucy, which models the lens/scan blur in linear light to pull back genuinely soft detail). Existing edits keep their look on Unsharp Mask. diff --git a/negpy/desktop/controller.py b/negpy/desktop/controller.py index a6cb0dcb..e432f06a 100644 --- a/negpy/desktop/controller.py +++ b/negpy/desktop/controller.py @@ -10,7 +10,7 @@ from negpy.desktop.converters import ImageConverter from negpy.desktop.render_memo import RenderMemo -from negpy.desktop.session import AppState, DesktopSessionManager, ToolMode, resolve_asset_rgbscan +from negpy.desktop.session import AppState, DesktopSessionManager, ToolMode, resolve_asset_rgbscan, resolve_asset_stitch from negpy.desktop.workers.export import ExportTask, ExportWorker, find_export_conflicts from negpy.desktop.workers.render import ( AssetDiscoveryTask, @@ -29,6 +29,8 @@ ThumbnailWorker, ) from negpy.desktop.workers.scan_worker import BatchRequest, RollPreviewRequest, ScanRequest, ScanWorker +from negpy.desktop.workers.stitch import StitchTask, StitchWorker +from negpy.features.stitch.models import stitch_hash, stitch_name from negpy.desktop.workers.capture_worker import ( CalibrationRequest, CaptureRequest, @@ -127,6 +129,7 @@ class _DiscoveryRequest: reselect_path: Optional[str] rgb_scan: bool half_frame: bool + restore_stitches: Optional[dict] = None def baseline_compare_config(config: WorkspaceConfig) -> WorkspaceConfig: @@ -176,6 +179,7 @@ class AppController(QObject): crop_guide_changed = pyqtSignal() dust_overlay_changed = pyqtSignal() asset_discovery_requested = pyqtSignal(AssetDiscoveryTask) + stitch_requested = pyqtSignal(object) thumbnail_requested = pyqtSignal(list) thumbnail_update_requested = pyqtSignal(ThumbnailUpdateTask) tool_sync_requested = pyqtSignal() @@ -281,6 +285,9 @@ def __init__(self, session_manager: DesktopSessionManager): self.export_thread = QThread() self.export_worker = ExportWorker() self.export_worker.moveToThread(self.export_thread) + # Shares the export thread: the batch lane serializes them anyway. + self.stitch_worker = StitchWorker() + self.stitch_worker.moveToThread(self.export_thread) self.export_thread.start() self.thumb_thread = QThread() @@ -445,6 +452,12 @@ def _connect_signals(self) -> None: self.export_worker.error.connect(self._on_render_error) self.export_worker.error.connect(self._on_export_task_error) + self.stitch_requested.connect(self.stitch_worker.run) + self.stitch_worker.progress.connect(self._on_batch_progress) + self.stitch_worker.registered.connect(self._on_stitch_registered) + self.stitch_worker.cancelled.connect(self._on_stitch_cancelled) + self.stitch_worker.error.connect(self._on_stitch_error) + self.thumbnail_requested.connect(self.thumb_worker.generate) self.thumb_worker.progress.connect(self._on_thumbnail_progress) self.thumbnail_update_requested.connect(self.thumb_worker.update_rendered) @@ -633,6 +646,8 @@ def abort_active_batch(self) -> None: elif self._active_batch == "autocrop": self._autocrop_cancel_requested = True self.batch_autocrop_worker.cancel(self._autocrop_batch_token) + elif self._active_batch == "stitch": + self.stitch_worker.cancel() def saved_session_paths(self) -> List[str]: """Returns last session's file paths that still exist on disk.""" @@ -647,7 +662,8 @@ def restore_session(self) -> None: active = self.session.repo.get_global_setting("session_active_path") self._pending_scanned_file = active if active in paths else paths[0] triplets = self.session.repo.get_global_setting("session_triplets", {}) or {} - self.request_asset_discovery(paths, auto_open=True, restore_triplets=triplets) + stitches = self.session.repo.get_global_setting("session_stitches", {}) or {} + self.request_asset_discovery(paths, auto_open=True, restore_triplets=triplets, restore_stitches=stitches) def request_asset_discovery( self, @@ -656,6 +672,7 @@ def request_asset_discovery( restore_triplets: Optional[dict] = None, replace_existing: bool = False, reselect_path: Optional[str] = None, + restore_stitches: Optional[dict] = None, ) -> None: """ Starts asynchronous discovery of supported assets. @@ -673,6 +690,7 @@ def request_asset_discovery( reselect_path=reselect_path, rgb_scan=bool(self.session.repo.get_global_setting("rgbscan_mode", False)), half_frame=bool(self.session.repo.get_global_setting("half_frame_mode", False)), + restore_stitches=restore_stitches, ) if self._discovery_running: self._pending_asset_discoveries.append(request) @@ -705,6 +723,7 @@ def _start_asset_discovery(self, request: _DiscoveryRequest) -> None: rgb_scan=request.rgb_scan, restore_triplets=request.restore_triplets, half_frame=request.half_frame, + restore_stitches=request.restore_stitches, ) self.asset_discovery_requested.emit(task) @@ -913,6 +932,8 @@ def load_file(self, file_path: str, preserve_zoom: bool = False, force_detect: b self.state.is_dirty = True rgbscan = self.state.config.rgbscan + stitch = self.state.config.stitch + flatfield = self.state.config.flatfield self.preview_load_requested.emit( PreviewLoadTask( file_path=file_path, @@ -931,6 +952,11 @@ def load_file(self, file_path: str, preserve_zoom: bool = False, force_detect: b green_path=rgbscan.green_path if rgbscan.enabled else "", blue_path=rgbscan.blue_path if rgbscan.enabled else "", align=rgbscan.align, + stitch_paths=stitch.stitch_paths if stitch.stitch_enabled else (), + stitch_transforms=stitch.stitch_transforms if stitch.stitch_enabled else (), + stitch_canvas=stitch.stitch_canvas, + stitch_sizes=stitch.stitch_sizes, + flatfield_path=flatfield.reference_path if (stitch.stitch_enabled and flatfield.apply) else "", ) ) @@ -1280,7 +1306,7 @@ def apply_auto_crop(self) -> None: def _config_for_autocrop_asset(self, asset: dict) -> WorkspaceConfig: """Resolve per-asset settings, including unsaved edits on the active frame.""" if asset.get("hash") == self.state.current_file_hash: - return resolve_asset_rgbscan(self.state.config, asset) + return resolve_asset_stitch(resolve_asset_rgbscan(self.state.config, asset), asset) return self.session.config_for_asset(asset) def request_batch_auto_crop(self) -> None: @@ -2068,6 +2094,76 @@ def _on_scan_batch_finished(self, paths: list) -> None: self._pending_scanned_file = paths[-1] self.request_asset_discovery(list(paths)) + # ── Stitch (multi-part scan composite) ───────────────────────────── + + def request_stitch_selected(self) -> None: + """Register the selected frames into one stitched composite asset.""" + if self._batch_busy("Stitch"): + return + files = [self.state.uploaded_files[i] for i in sorted(set(self.state.selected_indices)) if 0 <= i < len(self.state.uploaded_files)] + by_path = {f["path"]: f for f in files} # half-frame assets share a path + ordered = sorted(by_path.values(), key=lambda f: os.path.basename(f["path"]).lower()) + if len(ordered) < 2: + self.set_status("Select two or more frames to stitch", 4000) + return + if any(f.get("green_path") or f.get("stitch_paths") for f in ordered): + self.set_status("Stitching RGB-scan or already-stitched frames is not supported", 4000) + return + if self._begin_batch("stitch", "Stitching frames", abortable=True) is None: + return + self.stitch_requested.emit( + StitchTask( + files=tuple(dict(f) for f in ordered), + params_by_path={f["path"]: self._batch_params_for(f) for f in ordered}, + ) + ) + + def _on_stitch_registered(self, payload: dict) -> None: + self._end_batch("stitch") + files = payload["files"] + part_paths = [f["path"] for f in files] + composite = { + "name": stitch_name(part_paths), + "path": part_paths[0], + "hash": stitch_hash([f["hash"] for f in files]), + "stitch_paths": tuple(part_paths[1:]), + "stitch_transforms": payload["transforms"], + "stitch_canvas": payload["canvas"], + "stitch_sizes": payload["sizes"], + } + wanted = set(part_paths) + indices = [i for i, f in enumerate(self.state.uploaded_files) if f["path"] in wanted] + self.session.apply_stitch(indices, composite) + self.set_status(f"Stitched {len(files)} frames", 4000) + # The composite bypasses asset discovery, so nothing else queues its thumbnail. + self.generate_missing_thumbnails() + + def _on_stitch_cancelled(self) -> None: + self._on_batch_cancelled("stitch") + + def _on_stitch_error(self, message: str) -> None: + self._end_batch("stitch") + self.set_status(message, 6000) + + def request_unstitch(self) -> None: + """Dissolve the active stitched composite back into its part frames. + + Part edits restore from the DB by content hash; the composite's edits stay + keyed under its stitch hash for a future re-stitch of the same parts.""" + idx = self.state.selected_file_idx + if not (0 <= idx < len(self.state.uploaded_files)): + return + asset = self.state.uploaded_files[idx] + parts = asset.get("stitch_paths") + if not parts: + return + paths = [asset["path"], *parts] + self.state.uploaded_files.pop(idx) + self.session.state.thumbnails.pop(asset["name"], None) + self.session.asset_model.refresh() + self._pending_scanned_file = paths[0] + self.request_asset_discovery(paths) + def _select_file_by_path(self, path: str) -> bool: """Find a file by path in uploaded_files and select it.""" for i, f_info in enumerate(self.session.state.uploaded_files): @@ -2412,7 +2508,7 @@ def _batch_params_for(self, f: dict) -> WorkspaceConfig: config), with its own RGB-scan green/blue re-injected from the asset dict — the same authoritative source individual export gets via select_file.""" params = self.session.repo.load_file_settings(f["hash"]) or self.state.config - return resolve_asset_rgbscan(params, f) + return resolve_asset_stitch(resolve_asset_rgbscan(params, f), f) def _tasks_for_file( self, diff --git a/negpy/desktop/session.py b/negpy/desktop/session.py index 7d6aad13..1bf6f2e4 100644 --- a/negpy/desktop/session.py +++ b/negpy/desktop/session.py @@ -11,6 +11,7 @@ from negpy.domain.models import ExportPreset, WorkspaceConfig from negpy.features.exposure.models import apply_targets from negpy.features.rgbscan.models import RgbScanConfig +from negpy.features.stitch.models import StitchConfig from negpy.infrastructure.display.color_spaces import WORKING_COLOR_SPACE from negpy.infrastructure.storage.repository import StorageRepository from negpy.kernel.system.config import APP_CONFIG @@ -421,6 +422,27 @@ def resolve_asset_rgbscan(params: WorkspaceConfig, asset: dict) -> WorkspaceConf return replace(params, rgbscan=RgbScanConfig()) +def resolve_asset_stitch(params: WorkspaceConfig, asset: dict) -> WorkspaceConfig: + """Overlay a composite's stored registration (from the asset dict — the authoritative + source) onto its params. A non-stitch asset gets stitch reset so a plain frame never + inherits a leaked composite config. Session/JSON round-trips lists — coerce to tuples + so the frozen config stays hashable.""" + paths = asset.get("stitch_paths") + if paths: + canvas = asset.get("stitch_canvas") or (0, 0) + return replace( + params, + stitch=StitchConfig( + stitch_enabled=True, + stitch_paths=tuple(paths), + stitch_transforms=tuple(tuple(float(v) for v in t) for t in asset.get("stitch_transforms") or ()), + stitch_canvas=(int(canvas[0]), int(canvas[1])), + stitch_sizes=tuple((int(s[0]), int(s[1])) for s in asset.get("stitch_sizes") or ()), + ), + ) + return replace(params, stitch=StitchConfig()) + + class DesktopSessionManager(QObject): """ Manages application state, file list, and configuration persistence. @@ -776,7 +798,7 @@ def _hydrate_asset_config(self, asset: dict) -> tuple[WorkspaceConfig, bool]: config = self._apply_sticky_settings(saved_config, only_global=True) else: config = self._apply_sticky_settings(WorkspaceConfig(), only_global=False) - return resolve_asset_rgbscan(config, asset), is_new + return resolve_asset_stitch(resolve_asset_rgbscan(config, asset), asset), is_new def config_for_asset(self, asset: dict) -> WorkspaceConfig: """Return an asset's hydrated config without changing the active session state. @@ -1101,6 +1123,20 @@ def _persist_session(self) -> None: if f.get("green_path") and f.get("blue_path") } self.repo.save_global_setting("session_triplets", triplets) + # Stitch composites keep their parts + registration here so restore can rebuild + # the merged asset without re-running SIFT (re-discovery sees only the primary). + stitches = { + f["path"]: { + "paths": list(f["stitch_paths"]), + "transforms": [list(t) for t in f["stitch_transforms"]], + "canvas": list(f["stitch_canvas"]), + "sizes": [list(s) for s in f["stitch_sizes"]], + "hash": f["hash"], + } + for f in self.state.uploaded_files + if f.get("stitch_paths") + } + self.repo.save_global_setting("session_stitches", stitches) def add_files(self, file_paths: List[str], validated_info: Optional[List[Dict]] = None) -> None: """ @@ -1157,6 +1193,26 @@ def add_files(self, file_paths: List[str], validated_info: Optional[List[Dict]] self.files_changed.emit() self._persist_session() + def apply_stitch(self, indices: List[int], composite: dict) -> None: + """Replace the part assets with their stitched composite (inserted at the first + part's position), then open it. Part edits stay in the DB under their content + hashes, so an unstitch restores them intact.""" + valid = sorted({i for i in indices if 0 <= i < len(self.state.uploaded_files)}) + if not valid: + return + pos = valid[0] + for i in reversed(valid): + removed = self.state.uploaded_files.pop(i) + self.state.thumbnails.pop(removed["name"], None) + marks = self.repo.load_file_marks() + m = marks.get(composite["hash"]) + composite = {**composite, "keeper": m == "keeper", "excluded": m == "excluded"} + self.state.uploaded_files.insert(pos, composite) + self.asset_model.refresh() + self.files_changed.emit() + self._persist_session() + self.select_file(pos) + def set_triplet(self, index: int, red_path: str, green_path: str, blue_path: str, align: bool = True) -> None: """Reassign the R/G/B exposures of an RGB-scan asset, then reload it.""" import os diff --git a/negpy/desktop/view/sidebar/files.py b/negpy/desktop/view/sidebar/files.py index 4517fe79..74f952a6 100644 --- a/negpy/desktop/view/sidebar/files.py +++ b/negpy/desktop/view/sidebar/files.py @@ -764,9 +764,15 @@ def _build_context_menu(self) -> QMenu: act_reject.triggered.connect(lambda: self.session.toggle_mark("excluded")) menu.addSeparator() menu.addAction("Apply settings…").triggered.connect(self._open_apply_dialog) - if not multi: + if multi: + menu.addSeparator() + menu.addAction("Stitch selected frames").triggered.connect(lambda: self.controller.request_stitch_selected()) + else: menu.addSeparator() menu.addAction("Edit RGB Triplet…").triggered.connect(self._on_edit_triplet) + active = state.uploaded_files[state.selected_file_idx] if 0 <= state.selected_file_idx < len(state.uploaded_files) else {} + if active.get("stitch_paths"): + menu.addAction("Unstitch").triggered.connect(lambda: self.controller.request_unstitch()) menu.addSeparator() unload_label = "Unload Selected" if multi else "Unload" menu.addAction(unload_label).triggered.connect(self._on_remove_from_menu) diff --git a/negpy/desktop/workers/render.py b/negpy/desktop/workers/render.py index 492c5265..c76f4ece 100644 --- a/negpy/desktop/workers/render.py +++ b/negpy/desktop/workers/render.py @@ -118,6 +118,7 @@ class AssetDiscoveryTask: rgb_scan: bool = False # Group discovered files into R/G/B triplets (one asset per frame). restore_triplets: dict | None = None # {red_path: [green, blue]} — rebuild known triplets (session restore). half_frame: bool = False # Expand each file into two half-frame assets (left/right). + restore_stitches: dict | None = None # {primary_path: {paths, transforms, canvas, sizes, hash}} (session restore). @dataclass(frozen=True) @@ -135,6 +136,11 @@ class PreviewLoadTask: green_path: str = "" # RGB-scan triplet: green/blue exposures merged with file_path (red). blue_path: str = "" align: bool = True # sub-pixel registration of the triplet + stitch_paths: tuple[str, ...] = () # stitch composite: non-primary parts + stored registration + stitch_transforms: tuple[tuple[float, ...], ...] = () + stitch_canvas: tuple[int, int] = (0, 0) + stitch_sizes: tuple[tuple[int, int], ...] = () + flatfield_path: str = "" # per-part flat-field for stitch previews class RenderWorker(QObject): @@ -334,6 +340,9 @@ def process(self, task: AssetDiscoveryTask) -> None: elif task.rgb_scan and valid_assets: valid_assets = self._group_rgb_triplets(valid_assets) + if task.restore_stitches and valid_assets: + valid_assets = self._attach_restored_stitches(valid_assets, task.restore_stitches) + if task.half_frame and valid_assets: valid_assets = self._expand_half_frames(valid_assets) @@ -346,7 +355,7 @@ def _expand_half_frames(self, assets: list) -> list: out = [] for i, a in enumerate(assets): - if a.get("green_path"): + if a.get("green_path") or a.get("stitch_paths"): out.append(a) continue self.progress.emit(i + 1, len(assets), f"Split {a['name']}") @@ -370,6 +379,32 @@ def _attach_restored_triplets(self, assets: list, triplets: dict) -> list: out.append(a) return out + def _attach_restored_stitches(self, assets: list, stitches: dict) -> list: + """Re-attach saved stitch registrations to restored primary assets (no re-registration). + A composite whose parts vanished from disk restores as a plain asset.""" + import os + + from negpy.features.stitch.models import stitch_name + + out = [] + for a in assets: + entry = stitches.get(a["path"]) + if entry and entry.get("paths") and all(os.path.exists(p) for p in entry["paths"]): + out.append( + { + **a, + "name": stitch_name([a["path"], *entry["paths"]]), + "hash": entry["hash"], + "stitch_paths": tuple(entry["paths"]), + "stitch_transforms": tuple(tuple(float(v) for v in t) for t in entry["transforms"]), + "stitch_canvas": (int(entry["canvas"][0]), int(entry["canvas"][1])), + "stitch_sizes": tuple((int(s[0]), int(s[1])) for s in entry["sizes"]), + } + ) + else: + out.append(a) + return out + def _group_rgb_triplets(self, assets: list) -> list: """Classify each file by dominant channel and merge consecutive R/G/B triplets into one asset (red is primary; green/blue ride along). Unmatched files stay individual.""" @@ -435,6 +470,37 @@ def process(self, task: PreviewLoadTask) -> None: return t0 = time.perf_counter() try: + if task.stitch_paths: + # Stitch composite: replay the stored registration at preview scale. + # No splash — the primary's embedded JPEG would flash a half frame. + from negpy.features.stitch.models import StitchConfig + + stitch_cfg = StitchConfig( + stitch_enabled=True, + stitch_paths=task.stitch_paths, + stitch_transforms=task.stitch_transforms, + stitch_canvas=task.stitch_canvas, + stitch_sizes=task.stitch_sizes, + ) + raw, dims, metadata = self._preview_service.load_linear_preview_stitch( + task.file_path, + stitch_cfg, + task.workspace_color_space, + use_camera_wb=task.use_camera_wb, + full_resolution=task.full_resolution, + file_hash=task.file_hash, + flatfield_path=task.flatfield_path, + ) + source_cs = metadata.get("color_space") or WORKING_COLOR_SPACE + ir_preview = metadata.get("ir_preview") + detected_mode = self._detect_mode(task, raw) if task.detect_mode else "" + logger.info( + "load-timing preview_worker_total %.0fms (stitch load->buffer) %s", + (time.perf_counter() - t0) * 1000, + task.file_path, + ) + self.finished.emit(task.file_path, raw, dims, source_cs, ir_preview, detected_mode) + return if task.green_path and task.blue_path: # RGB-scan triplet: assemble the frame from the three exposures. # No splash — the red embedded JPEG would flash a red-cast preview. diff --git a/negpy/desktop/workers/stitch.py b/negpy/desktop/workers/stitch.py new file mode 100644 index 00000000..87e8c2e3 --- /dev/null +++ b/negpy/desktop/workers/stitch.py @@ -0,0 +1,80 @@ +import gc +import threading +from dataclasses import dataclass +from typing import Dict, Tuple + +from PyQt6.QtCore import QObject, pyqtSignal, pyqtSlot + +from negpy.domain.models import WorkspaceConfig +from negpy.features.stitch.logic import StitchCancelled, StitchError, register_parts +from negpy.services.rendering.image_processor import ImageProcessor + + +@dataclass(frozen=True) +class StitchTask: + """Immutable request to register a set of overlapping scans. + + ``files`` are asset-dict copies in registration order; files[0] is the + primary/reference part. Params carry each part's decode-relevant settings + (flat-field, linear-raw) so registration sees the same buffers a later + decode replays the transforms against. + """ + + files: Tuple[dict, ...] + params_by_path: Dict[str, WorkspaceConfig] + + +class StitchWorker(QObject): + """Registers the parts full-res and emits the resulting geometry. + + No file writes and no full-res blend: registration success is the gate; + the composite itself is assembled at decode/preview time from the payload. + """ + + progress = pyqtSignal(int, int, str) # current, total, label + registered = pyqtSignal(object) # {"files", "transforms", "canvas", "sizes"} + cancelled = pyqtSignal() + error = pyqtSignal(str) + + def __init__(self) -> None: + super().__init__() + self._processor = ImageProcessor() + self._cancel = threading.Event() + + @pyqtSlot() + def cancel(self) -> None: + self._cancel.set() + + @pyqtSlot(object) + def run(self, task: StitchTask) -> None: + self._cancel.clear() + parts = [] + total = len(task.files) + 1 + try: + for i, f in enumerate(task.files): + if self._cancel.is_set(): + self.cancelled.emit() + return + self.progress.emit(i, total, f"Decoding {f['name']}") + f32, _, _ = self._processor._decode_oriented_f32(f["path"], task.params_by_path[f["path"]]) + parts.append(f32) + self.progress.emit(len(task.files), total, "Registering frames") + transforms, canvas = register_parts(parts, is_cancelled=self._cancel.is_set) + self.registered.emit( + { + "files": list(task.files), + "transforms": tuple(tuple(float(v) for v in m.ravel()) for m in transforms), + "canvas": (int(canvas[0]), int(canvas[1])), + "sizes": tuple((p.shape[1], p.shape[0]) for p in parts), + } + ) + except StitchCancelled: + self.cancelled.emit() + except StitchError as e: + self.error.emit(str(e)) + except Exception as e: + self.error.emit(f"Stitch failed: {e}") + finally: + parts.clear() + self._processor.cleanup(release_source_cache=True, collect=False) + gc.collect() diff --git a/negpy/domain/models.py b/negpy/domain/models.py index 905631eb..a4bc532c 100644 --- a/negpy/domain/models.py +++ b/negpy/domain/models.py @@ -15,6 +15,7 @@ from negpy.features.finish.models import FinishConfig from negpy.features.flatfield.models import FlatFieldConfig from negpy.features.rgbscan.models import RgbScanConfig +from negpy.features.stitch.models import StitchConfig from negpy.features.metadata.models import MetadataConfig from negpy.kernel.system.logging import get_logger import negpy.kernel.system.paths as paths @@ -361,6 +362,7 @@ class WorkspaceConfig: exposure: ExposureConfig = field(default_factory=ExposureConfig) flatfield: FlatFieldConfig = field(default_factory=FlatFieldConfig) rgbscan: RgbScanConfig = field(default_factory=RgbScanConfig) + stitch: StitchConfig = field(default_factory=StitchConfig) geometry: GeometryConfig = field(default_factory=GeometryConfig) lab: LabConfig = field(default_factory=LabConfig) local: LocalAdjustmentsConfig = field(default_factory=LocalAdjustmentsConfig) @@ -379,6 +381,7 @@ def to_dict(self) -> Dict[str, Any]: res.update(asdict(self.exposure)) res.update(asdict(self.flatfield)) res.update(asdict(self.rgbscan)) + res.update(asdict(self.stitch)) res.update(asdict(self.geometry)) res.update(asdict(self.lab)) res["local_masks"] = asdict(self.local) @@ -514,11 +517,24 @@ def _build_local(d: Dict[str, Any]) -> LocalAdjustmentsConfig: ) return LocalAdjustmentsConfig(masks=tuple(masks)) + def _build_stitch(d: Dict[str, Any]) -> StitchConfig: + # JSON round-trips tuples as lists; coerce back or the frozen config + # loses hashability/equality (breaks config-diff caching). + canvas = d.get("stitch_canvas", (0, 0)) + return StitchConfig( + stitch_enabled=bool(d.get("stitch_enabled", False)), + stitch_paths=tuple(d.get("stitch_paths", ())), + stitch_transforms=tuple(tuple(float(v) for v in row) for row in d.get("stitch_transforms", ())), + stitch_canvas=(int(canvas[0]), int(canvas[1])), + stitch_sizes=tuple((int(s[0]), int(s[1])) for s in d.get("stitch_sizes", ())), + ) + return cls( process=ProcessConfig(**filter_keys(ProcessConfig, data)), exposure=ExposureConfig(**filter_keys(ExposureConfig, data)), flatfield=FlatFieldConfig(**filter_keys(FlatFieldConfig, data)), rgbscan=RgbScanConfig(**filter_keys(RgbScanConfig, data)), + stitch=_build_stitch(filter_keys(StitchConfig, data)), geometry=GeometryConfig(**filter_keys(GeometryConfig, data)), lab=LabConfig(**filter_keys(LabConfig, data)), local=_build_local(local_data), diff --git a/negpy/features/stitch/__init__.py b/negpy/features/stitch/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/negpy/features/stitch/logic.py b/negpy/features/stitch/logic.py new file mode 100644 index 00000000..3a9fd5d1 --- /dev/null +++ b/negpy/features/stitch/logic.py @@ -0,0 +1,222 @@ +"""Registration and blending for multi-part scan stitching. + +All functions operate on scene-linear float32 buffers that are already +EXIF-oriented and flat-fielded — transforms estimated here are only valid on +buffers decoded the same way. +""" + +from typing import Callable, List, Optional, Sequence, Tuple + +import cv2 +import numpy as np + +from negpy.features.stitch.models import StitchConfig + +PROXY_MAX_EDGE = 1600 +MIN_INLIERS = 100 + +_NO_OVERLAP_MSG = "Could not find overlap between the selected frames" + + +class StitchError(RuntimeError): + """Registration failure; the message is shown to the user verbatim.""" + + +class StitchCancelled(StitchError): + pass + + +def build_proxy(rgb: np.ndarray) -> Tuple[np.ndarray, float]: + """Registration proxy: green channel, <=1600px, contrast-stretched uint8. + + Returns (proxy, scale) with proxy_coords = full_coords * scale. + """ + gray = rgb[..., 1] if rgb.ndim == 3 else rgb + scale = min(1.0, PROXY_MAX_EDGE / max(gray.shape)) + if scale < 1.0: + size = (max(1, round(gray.shape[1] * scale)), max(1, round(gray.shape[0] * scale))) + gray = cv2.resize(gray, size, interpolation=cv2.INTER_AREA) + lo, hi = np.percentile(gray, (1.0, 99.0)) + gray = np.clip((gray - lo) / max(hi - lo, 1e-9), 0.0, 1.0) + return (gray ** (1.0 / 2.2) * 255.0).astype(np.uint8), scale + + +def register_pair(ref_proxy: np.ndarray, mov_proxy: np.ndarray, min_inliers: int = MIN_INLIERS) -> Tuple[np.ndarray, int]: + """Similarity transform (2x3) mapping ``mov_proxy`` coords to ``ref_proxy`` coords. + + SIFT + ratio test + RANSAC; ORB and phase correlation both fail on film + negatives (feature-poor orange-mask content, sub-degree rotations). + """ + sift = cv2.SIFT_create(nfeatures=6000) + ref_kp, ref_desc = sift.detectAndCompute(ref_proxy, None) + mov_kp, mov_desc = sift.detectAndCompute(mov_proxy, None) + if ref_desc is None or mov_desc is None or len(ref_kp) < 2 or len(mov_kp) < 2: + raise StitchError(_NO_OVERLAP_MSG) + + pairs = cv2.BFMatcher().knnMatch(mov_desc, ref_desc, k=2) + good = [m for m, n in (p for p in pairs if len(p) == 2) if m.distance < 0.75 * n.distance] + if len(good) < min_inliers: + raise StitchError(_NO_OVERLAP_MSG) + + mov_pts = np.float32([mov_kp[m.queryIdx].pt for m in good]) + ref_pts = np.float32([ref_kp[m.trainIdx].pt for m in good]) + matrix, inlier_mask = cv2.estimateAffinePartial2D(mov_pts, ref_pts, method=cv2.RANSAC, ransacReprojThreshold=3.0) + inliers = int(inlier_mask.sum()) if inlier_mask is not None else 0 + if matrix is None or inliers < min_inliers: + raise StitchError(_NO_OVERLAP_MSG) + return matrix.astype(np.float64), inliers + + +def scale_affine(matrix: np.ndarray, ref_scale: float, mov_scale: float) -> np.ndarray: + """Proxy-coordinate affine -> full-resolution affine.""" + out = matrix.astype(np.float64).copy() + out[:, :2] *= mov_scale / ref_scale + out[:, 2] /= ref_scale + return out + + +def chain_affine(a: np.ndarray, b: np.ndarray) -> np.ndarray: + """Compose 2x3 affines: (a ∘ b)(x) = a(b(x)).""" + a3 = np.vstack([a, (0.0, 0.0, 1.0)]) + b3 = np.vstack([b, (0.0, 0.0, 1.0)]) + return (a3 @ b3)[:2] + + +def compute_canvas(sizes: Sequence[Tuple[int, int]], transforms: Sequence[np.ndarray]) -> Tuple[Tuple[int, int], List[np.ndarray]]: + """Union bounding box of the warped parts. Returns ((W, H), transforms shifted to it).""" + corners = [] + for (w, h), matrix in zip(sizes, transforms): + box = np.array([[[0, 0], [w, 0], [w, h], [0, h]]], np.float64) + corners.append(cv2.transform(box, matrix)[0]) + points = np.vstack(corners) + lo = np.floor(points.min(axis=0)) + hi = np.ceil(points.max(axis=0)) + shift = np.array([[1.0, 0.0, -lo[0]], [0.0, 1.0, -lo[1]]], np.float64) + shifted = [chain_affine(shift, m) for m in transforms] + return (int(hi[0] - lo[0]), int(hi[1] - lo[1])), shifted + + +def register_parts( + parts: Sequence[np.ndarray], + is_cancelled: Callable[[], bool] = lambda: False, + on_progress: Callable[[int, int, str], None] = lambda *a: None, +) -> Tuple[List[np.ndarray], Tuple[int, int]]: + """Chained pairwise registration (each part vs the previous — adjacent shots overlap). + + Returns full-res part->canvas transforms (incl. the identity-based primary) + and the canvas size. Raises StitchError when a pair has no usable overlap. + """ + proxies = [build_proxy(p) for p in parts] + transforms = [np.array([[1.0, 0.0, 0.0], [0.0, 1.0, 0.0]], np.float64)] + for i in range(1, len(parts)): + if is_cancelled(): + raise StitchCancelled("Cancelled") + on_progress(i, len(parts) - 1, f"Registering frame {i + 1}") + matrix, _ = register_pair(proxies[i - 1][0], proxies[i][0]) + full = scale_affine(matrix, proxies[i - 1][1], proxies[i][1]) + transforms.append(chain_affine(transforms[i - 1], full)) + sizes = [(p.shape[1], p.shape[0]) for p in parts] + canvas, shifted = compute_canvas(sizes, transforms) + return shifted, canvas + + +def scale_transforms(config: StitchConfig, decoded_sizes: Sequence[Tuple[int, int]]) -> Tuple[List[np.ndarray], Tuple[int, int]]: + """Stored full-res transforms -> the scale the parts were actually decoded at. + + Per-part scale comes from ``stitch_sizes`` (preview decodes can round each + part differently); canvas coordinates scale with the primary part. + """ + ref_scale = decoded_sizes[0][0] / config.stitch_sizes[0][0] + out = [] + for flat, (dec_w, _), (full_w, _) in zip(config.stitch_transforms, decoded_sizes, config.stitch_sizes): + matrix = np.array(flat, np.float64).reshape(2, 3) + part_scale = dec_w / full_w + matrix[:, :2] *= ref_scale / part_scale + matrix[:, 2] *= ref_scale + out.append(matrix) + canvas = (round(config.stitch_canvas[0] * ref_scale), round(config.stitch_canvas[1] * ref_scale)) + return out, canvas + + +def warp_into_canvas( + img: np.ndarray, + transform: np.ndarray, + canvas_wh: Tuple[int, int], + interpolation: int = cv2.INTER_CUBIC, +) -> Tuple[np.ndarray, np.ndarray]: + """Warp a part and its validity mask into the canvas. The mask is eroded by the + >0.999 threshold so resampled border pixels never bleed into the blend.""" + warped = cv2.warpAffine(img, transform.astype(np.float32), canvas_wh, flags=interpolation) + ones = np.ones(img.shape[:2], np.float32) + mask = cv2.warpAffine(ones, transform.astype(np.float32), canvas_wh, flags=cv2.INTER_LINEAR) + return warped, mask > 0.999 + + +_MIN_OVERLAP_PX = 1000 + + +def gain_compensate(ref: np.ndarray, ref_mask: np.ndarray, mov: np.ndarray, mov_mask: np.ndarray) -> np.ndarray: + """Scale ``mov`` (in place) so its overlap with ``ref`` matches per channel — + removes light-source drift between shots. No-op on tiny overlap.""" + overlap = ref_mask & mov_mask + if int(overlap.sum()) < _MIN_OVERLAP_PX: + return mov + ref_mean = ref[overlap].mean(axis=0) + mov_mean = mov[overlap].mean(axis=0) + if np.any(mov_mean < 1e-6): + return mov + mov *= (ref_mean / mov_mean).astype(np.float32) + return mov + + +def feather_blend(warped: Sequence[np.ndarray], masks: Sequence[np.ndarray]) -> np.ndarray: + """Distance-transform-weighted average: seams fade over the full overlap width.""" + acc = np.zeros_like(warped[0]) + weight_sum = np.zeros(warped[0].shape[:2], np.float32) + for img, mask in zip(warped, masks): + weight = cv2.distanceTransform(mask.astype(np.uint8), cv2.DIST_L2, 3) + acc += img * weight[..., None] + weight_sum += weight + return acc / np.maximum(weight_sum, 1e-6)[..., None] + + +def blend_ir(irs: Sequence[np.ndarray], transforms: Sequence[np.ndarray], canvas_wh: Tuple[int, int]) -> np.ndarray: + """Per-pixel max over valid contributions: film dust appears in every part and + survives; single-part (sensor-side) specks are erased. Uncovered pixels are 1.0 + (loader convention: clean).""" + w, h = canvas_wh + acc = np.full((h, w), -1.0, np.float32) + for ir, transform in zip(irs, transforms): + warped, mask = warp_into_canvas(ir, transform, canvas_wh, interpolation=cv2.INTER_LINEAR) + acc[mask] = np.maximum(acc[mask], warped[mask]) + acc[acc < 0.0] = 1.0 + return acc + + +def stitch_composite( + parts: List[np.ndarray], + irs: Sequence[Optional[np.ndarray]], + config: StitchConfig, +) -> Tuple[np.ndarray, Optional[np.ndarray]]: + """Warp + gain-compensate + feather-blend the decoded parts into one frame. + + ``parts`` is consumed (slots dropped after warping) to cap peak memory. + IR is carried only when every part has one. + """ + transforms, canvas = scale_transforms(config, [(p.shape[1], p.shape[0]) for p in parts]) + warped: List[np.ndarray] = [] + masks: List[np.ndarray] = [] + for i in range(len(parts)): + img, mask = warp_into_canvas(parts[i], transforms[i], canvas) + parts[i] = None # type: ignore[call-overload] + for j in range(len(warped)): + if (masks[j] & mask).sum() >= _MIN_OVERLAP_PX: + gain_compensate(warped[j], masks[j], img, mask) + break + warped.append(img) + masks.append(mask) + rgb = np.clip(feather_blend(warped, masks), 0.0, 1.0) + ir = None + if irs and all(x is not None for x in irs): + ir = blend_ir(irs, transforms, canvas) # type: ignore[arg-type] + return rgb, ir diff --git a/negpy/features/stitch/models.py b/negpy/features/stitch/models.py new file mode 100644 index 00000000..ec9fc28d --- /dev/null +++ b/negpy/features/stitch/models.py @@ -0,0 +1,51 @@ +import hashlib +import os +from dataclasses import dataclass +from typing import Sequence + + +@dataclass(frozen=True) +class StitchConfig: + """Panorama stitch: one frame assembled from overlapping part scans. + + The primary part is the asset's own file; the remaining parts ride here. + Transforms are full-res part->canvas affines fixed once at registration + time; decode replays them at whatever scale it works at. + """ + + stitch_enabled: bool = False + stitch_paths: tuple[str, ...] = () # non-primary parts, registration order + stitch_transforms: tuple[tuple[float, ...], ...] = () # per part incl. primary, 2x3 row-major + stitch_canvas: tuple[int, int] = (0, 0) # full-res (W, H) + stitch_sizes: tuple[tuple[int, int], ...] = () # full-res decoded (W, H) per part + + +def stitch_token(config: StitchConfig) -> str: + """Identity of the active stitch, folded into the render source hash. Empty when inactive.""" + if not config.stitch_enabled or not config.stitch_paths: + return "" + parts = [] + for path in config.stitch_paths: + try: + parts.append(f"{path}:{os.path.getmtime(path)}") + except OSError: + return "" + geometry = repr((config.stitch_transforms, config.stitch_canvas, config.stitch_sizes)) + digest = hashlib.sha256(geometry.encode()).hexdigest()[:12] + return f"|stitch:{':'.join(parts)}:{digest}" + + +def stitch_name(part_paths: Sequence[str]) -> str: + """Display name of a composite: joined part stems, elided beyond three parts.""" + stems = [os.path.splitext(os.path.basename(p))[0] for p in part_paths] + if len(stems) > 3: + return f"{stems[0]} +{len(stems) - 1} (Stitch)" + return f"{'+'.join(stems)} (Stitch)" + + +def stitch_hash(part_hashes: Sequence[str]) -> str: + """Edit-key identity of a composite: content hashes of the parts, order-sensitive + (order defines the reference frame). The ``#`` suffix follows the half-frame + convention so ``base_hash`` strips to a plain digest.""" + digest = hashlib.sha256("|".join(part_hashes).encode()).hexdigest() + return f"{digest}#stitch" diff --git a/negpy/services/rendering/image_processor.py b/negpy/services/rendering/image_processor.py index 9109c79b..bf763548 100644 --- a/negpy/services/rendering/image_processor.py +++ b/negpy/services/rendering/image_processor.py @@ -36,6 +36,8 @@ route_ir_defects, ) from negpy.features.rgbscan.logic import merge_rgb_triplet, rgbscan_token +from negpy.features.stitch.logic import stitch_composite +from negpy.features.stitch.models import stitch_token from negpy.domain.interfaces import PipelineContext from negpy.services.rendering.engine import DarkroomEngine from negpy.services.rendering.gpu_engine import GPUEngine @@ -285,8 +287,10 @@ def run_pipeline( ``skip_flatfield``: the export CPU fallbacks pass an already-flat-fielded buffer. """ # Flat-field is a source pre-correction (before geometry/crop); folding its token - # into source_hash invalidates the engine cache when it changes. - if not skip_flatfield: + # into source_hash invalidates the engine cache when it changes. Stitch buffers + # arrive per-part flat-fielded (both decode paths) — correcting the composite + # canvas as one frame would stretch the gain map across the seam. + if not skip_flatfield and not settings.stitch.stitch_enabled: img = apply_flatfield(img, settings.flatfield) h_orig, w_cols = img.shape[:2] # Fold the buffer resolution into source_hash: toggling HQ re-decodes the same @@ -297,6 +301,7 @@ def run_pipeline( source_hash + flatfield_token(settings.flatfield) + rgbscan_token(settings.rgbscan) + + stitch_token(settings.stitch) + linear_raw_token(settings.process) + ir_bake_token(settings.retouch, ir_buffer is not None) ) @@ -429,9 +434,11 @@ def _load_source_f32( ) -> Tuple[np.ndarray, Optional[np.ndarray], str]: """Decode a source file to a flatfield-corrected, EXIF-oriented float32 buffer. + A stitch composite decodes every part and assembles them by replaying the + registration stored in ``params.stitch``. + Returns (f32_buffer, ir_buffer, source_color_space). """ - linear_raw = params.process.linear_raw rgbcfg = params.rgbscan is_triplet = bool(rgbcfg.enabled and rgbcfg.green_path and rgbcfg.blue_path) # Narrowband triplet channels don't survive half_size CFA binning. @@ -441,10 +448,48 @@ def _load_source_f32( mtime = os.path.getmtime(file_path) except OSError: mtime = 0.0 - cache_key = (file_path, mtime, linear_raw, rgbscan_token(params.rgbscan), flatfield_token(params.flatfield), fast_decode) + cache_key = ( + file_path, + mtime, + params.process.linear_raw, + rgbscan_token(params.rgbscan), + stitch_token(params.stitch), + flatfield_token(params.flatfield), + fast_decode, + ) if cache_key == self._source_cache_key and self._source_cache_value is not None: return self._source_cache_value + if params.stitch.stitch_enabled and params.stitch.stitch_paths: + parts, irs = [], [] + source_cs = WORKING_COLOR_SPACE + for i, path in enumerate((file_path, *params.stitch.stitch_paths)): + f32, ir, cs = self._decode_oriented_f32(path, params, fast_decode) + if i == 0: + source_cs = cs + parts.append(f32) + irs.append(ir) + f32_buffer, ir_full = stitch_composite(parts, irs, params.stitch) + result = (f32_buffer, ir_full, source_cs) + else: + result = self._decode_oriented_f32(file_path, params, fast_decode) + + self._source_cache_key = cache_key + self._source_cache_value = result + return result + + def _decode_oriented_f32( + self, file_path: str, params: WorkspaceConfig, fast_decode: bool = False + ) -> Tuple[np.ndarray, Optional[np.ndarray], str]: + """Single-file decode tail: sensor RGB -> float32 -> EXIF orientation -> flatfield. + + Stitch registration is estimated on buffers produced here, so any decode + the transforms are replayed against must come through here too. + """ + linear_raw = params.process.linear_raw + rgbcfg = params.rgbscan + is_triplet = bool(rgbcfg.enabled and rgbcfg.green_path and rgbcfg.blue_path) + rgb, metadata = self._decode_sensor_rgb(file_path, linear_raw, fast=fast_decode) # No embedded profile (scanner-raw linear, sensor-native RAW) → the buffer is # already in the working space, so "Same as Source" exports without converting. @@ -479,10 +524,7 @@ def _decode(path: str) -> np.ndarray: f32_buffer = apply_flatfield(f32_buffer, params.flatfield) if ir_full is not None: ir_full = apply_exif_orientation(ir_full, orientation) - result = (f32_buffer, ir_full, source_cs) - self._source_cache_key = cache_key - self._source_cache_value = result - return result + return f32_buffer, ir_full, source_cs @staticmethod def _slice_half_source( @@ -528,6 +570,7 @@ def process_export( source_hash + flatfield_token(params.flatfield) + rgbscan_token(params.rgbscan) + + stitch_token(params.stitch) + linear_raw_token(params.process) + ir_bake_token(params.retouch, ir_full is not None) ) @@ -751,6 +794,7 @@ def render_display_array( source_hash + flatfield_token(params.flatfield) + rgbscan_token(params.rgbscan) + + stitch_token(params.stitch) + linear_raw_token(params.process) + ir_bake_token(params.retouch, ir_full is not None) ) diff --git a/negpy/services/rendering/preview_manager.py b/negpy/services/rendering/preview_manager.py index 7a3d17d1..4b0b25a5 100644 --- a/negpy/services/rendering/preview_manager.py +++ b/negpy/services/rendering/preview_manager.py @@ -16,8 +16,12 @@ from negpy.kernel.image.logic import apply_exif_orientation, ensure_rgb, uint16_to_float32 from negpy.kernel.image.validation import ensure_image from negpy.kernel.system.config import APP_CONFIG +from negpy.features.flatfield.logic import apply_flatfield, flatfield_token +from negpy.features.flatfield.models import FlatFieldConfig from negpy.features.retouch.logic import downsample_ir from negpy.features.rgbscan.logic import assemble_rgb +from negpy.features.stitch.logic import stitch_composite +from negpy.features.stitch.models import StitchConfig, stitch_token from negpy.kernel.system.logging import get_logger from negpy.services.rendering.preview_cache import PreviewBufferCache, PreviewCacheKey @@ -382,6 +386,56 @@ def _match(buf: ImageBuffer) -> np.ndarray: self._cache.put(merged_key, out, dims, dict(meta)) return out, dims, meta + def load_linear_preview_stitch( + self, + primary_path: str, + stitch: StitchConfig, + color_space: str | None = None, + use_camera_wb: bool = False, + full_resolution: bool = False, + file_hash: str | None = None, + flatfield_path: str = "", + ) -> Tuple[ImageBuffer, Dimensions, dict]: + """Assemble a stitch composite at preview scale by replaying the stored + registration. Flat-field is applied per part here (a composite canvas must + never be flat-fielded as one frame), so the pipeline skips its own step. + + Returned dims are the full-resolution canvas, matching the single-file + convention of (original height, width) alongside a downsampled buffer. + """ + flatfield = FlatFieldConfig(apply=bool(flatfield_path), reference_path=flatfield_path) + key = None + if file_hash and color_space is not None: + token = stitch_token(stitch) + if token: + key = PreviewCacheKey( + file_hash=f"stitch|{file_hash}|{token}{flatfield_token(flatfield)}", + use_camera_wb=use_camera_wb, + workspace_color_space=color_space, + full_resolution=full_resolution, + ) + hit = self._cache.get(key) + if hit is not None: + return hit # cache hit — caller must not mutate this buffer + + parts, irs = [], [] + meta: dict = {} + for i, path in enumerate((primary_path, *stitch.stitch_paths)): + # file_hash=None: the composite hash is not the parts' content hash. + out, _, part_meta = self.load_linear_preview(path, color_space, use_camera_wb, full_resolution, None) + parts.append(apply_flatfield(np.asarray(out, dtype=np.float32), flatfield)) + irs.append(part_meta.get("ir_preview")) + if i == 0: + meta = dict(part_meta) + + rgb, ir = stitch_composite(parts, irs, stitch) + meta["ir_preview"] = ir + out_buf = ensure_image(rgb) + dims = (stitch.stitch_canvas[1], stitch.stitch_canvas[0]) + if key is not None: + self._cache.put(key, out_buf, dims, meta) + return out_buf, dims, meta + def load_splash_and_linear( self, file_path: str, diff --git a/tests/test_controller.py b/tests/test_controller.py index 415ed9f5..d00d014f 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -1137,7 +1137,9 @@ def test_restore_session_selects_active_and_discovers(self): self._mock_settings([a.name, b.name], b.name) self.controller.restore_session() self.assertEqual(self.controller._pending_scanned_file, b.name) - self.controller.request_asset_discovery.assert_called_once_with([a.name, b.name], auto_open=True, restore_triplets={}) + self.controller.request_asset_discovery.assert_called_once_with( + [a.name, b.name], auto_open=True, restore_triplets={}, restore_stitches={} + ) def test_restore_session_no_saved_files_is_noop(self): self._mock_settings([], None) diff --git a/tests/test_stitch.py b/tests/test_stitch.py new file mode 100644 index 00000000..ac145edb --- /dev/null +++ b/tests/test_stitch.py @@ -0,0 +1,480 @@ +import numpy as np +import pytest + +import cv2 + +from negpy.features.stitch.logic import ( + StitchError, + build_proxy, + register_pair, + register_parts, + scale_affine, + scale_transforms, + stitch_composite, +) +from negpy.features.stitch.models import StitchConfig, stitch_hash, stitch_token + +_H, _W = 900, 2000 +_P0_W, _P1_W = 1200, 1280 + + +def _scene(h=_H, w=_W, seed=7): + """Textured scene with enough detail for SIFT to lock onto.""" + rng = np.random.default_rng(seed) + base = cv2.GaussianBlur(rng.random((h, w), dtype=np.float32), (0, 0), 1.5) + grad = np.linspace(0.0, 0.3, w, dtype=np.float32)[None, :] + mono = np.clip(0.1 + 0.6 * base + grad, 0.0, 1.0) + scene = np.stack([mono, mono * 0.9, mono * 1.1], axis=-1) + return np.clip(scene, 0.0, 1.0).astype(np.float32) + + +def _true_affine(tx=720.0, ty=6.0, angle_deg=0.4, scale=0.998): + """Ground truth mapping part1 coords -> scene (== part0) coords.""" + a = np.deg2rad(angle_deg) + c, s = scale * np.cos(a), scale * np.sin(a) + return np.array([[c, -s, tx], [s, c, ty]], dtype=np.float64) + + +def _parts(scene=None): + scene = _scene() if scene is None else scene + part0 = scene[:, :_P0_W].copy() + t = _true_affine() + part1 = cv2.warpAffine( + scene, + t.astype(np.float32), + (_P1_W, _H), + flags=cv2.INTER_LINEAR | cv2.WARP_INVERSE_MAP, + ) + return scene, part0, part1, t + + +def test_register_pair_recovers_known_affine(): + _, part0, part1, t = _parts() + p0, s0 = build_proxy(part0) + p1, s1 = build_proxy(part1) + m_proxy, inliers = register_pair(p0, p1) + assert inliers > 100 + m = scale_affine(m_proxy, s0, s1) + ang_est = np.degrees(np.arctan2(m[1, 0], m[0, 0])) + ang_true = np.degrees(np.arctan2(t[1, 0], t[0, 0])) + assert abs(ang_est - ang_true) < 0.05 + assert abs(np.hypot(m[0, 0], m[1, 0]) - 0.998) < 0.002 + assert abs(m[0, 2] - t[0, 2]) < 2.0 + assert abs(m[1, 2] - t[1, 2]) < 2.0 + + +def test_register_pair_unrelated_raises(): + rng = np.random.default_rng(1) + a = (rng.random((400, 400)) * 255).astype(np.uint8) + b = (rng.random((400, 400)) * 255).astype(np.uint8) + with pytest.raises(StitchError): + register_pair(a, b) + + +def _registered_config(part0, part1): + transforms, canvas = register_parts([part0, part1]) + return StitchConfig( + stitch_enabled=True, + stitch_paths=("/p1",), + stitch_transforms=tuple(tuple(float(v) for v in m.ravel()) for m in transforms), + stitch_canvas=canvas, + stitch_sizes=((part0.shape[1], part0.shape[0]), (part1.shape[1], part1.shape[0])), + ) + + +def test_register_parts_and_composite_seam(): + scene, part0, part1, _ = _parts() + cfg = _registered_config(part0, part1) + cw, ch = cfg.stitch_canvas + assert abs(cw - _W) < 20 + assert abs(ch - _H) < 20 + + rgb, ir = stitch_composite([part0, part1], [None, None], cfg) + assert ir is None + assert rgb.shape[:2] == (ch, cw) + # Part0 is the reference: canvas offset is its (pure translation) transform. + m0 = np.array(cfg.stitch_transforms[0], dtype=np.float64).reshape(2, 3) + ox, oy = int(round(m0[0, 2])), int(round(m0[1, 2])) + inner = (slice(oy + 20, oy + _H - 20), slice(ox + 20, ox + _W - 20)) + scene_inner = scene[20:-20, 20:-20] + rms = float(np.sqrt(np.mean((rgb[inner] - scene_inner) ** 2))) + assert rms < 0.02 + + +def test_gain_compensation_removes_cast(): + scene, part0, part1, _ = _parts() + cast = np.array([1.02, 0.98, 1.01], dtype=np.float32) + cfg = _registered_config(part0, part1 * cast) + rgb, _ = stitch_composite([part0, np.clip(part1 * cast, 0, 1)], [None, None], cfg) + m0 = np.array(cfg.stitch_transforms[0], dtype=np.float64).reshape(2, 3) + ox, oy = int(round(m0[0, 2])), int(round(m0[1, 2])) + inner = (slice(oy + 20, oy + _H - 20), slice(ox + 20, ox + _W - 20)) + rms = float(np.sqrt(np.mean((rgb[inner] - scene[20:-20, 20:-20]) ** 2))) + assert rms < 0.02 + + +def test_ir_blended_only_when_all_present(): + _, part0, part1, t = _parts() + cfg = _registered_config(part0, part1) + + rgb, ir = stitch_composite([part0, part1], [np.ones(part0.shape[:2], np.float32), None], cfg) + assert ir is None + + ir0 = np.ones(part0.shape[:2], np.float32) + ir1 = np.ones(part1.shape[:2], np.float32) + # Shared defect (film dust, same scene spot in both parts) must survive the max-blend. + sx, sy = 800, 450 + inv = cv2.invertAffineTransform(t) + px, py = cv2.transform(np.array([[[sx, sy]]], np.float64), inv)[0, 0] + ir0[sy - 4 : sy + 4, sx - 4 : sx + 4] = 0.1 + ir1[int(py) - 6 : int(py) + 6, int(px) - 6 : int(px) + 6] = 0.1 + # Single-part defect (sensor dust) inside the overlap is erased by the other + # part's clean pixels under the max-blend. + ir0[440:460, 1000:1020] = 0.1 + + _, ir_out = stitch_composite([part0, part1], [ir0, ir1], cfg) + assert ir_out is not None + m0 = np.array(cfg.stitch_transforms[0], dtype=np.float64).reshape(2, 3) + ox, oy = int(round(m0[0, 2])), int(round(m0[1, 2])) + assert ir_out[oy + sy, ox + sx] < 0.5 + assert ir_out[oy + 450, ox + 1010] > 0.9 + # Canvas corners outside every part are clean (1.0) by convention. + assert ir_out[0, -1] > 0.99 + + +def test_scale_transforms_half_scale(): + _, part0, part1, _ = _parts() + cfg = _registered_config(part0, part1) + full, _ = stitch_composite([part0, part1], [None, None], cfg) + + halves = [cv2.resize(p, (p.shape[1] // 2, p.shape[0] // 2), interpolation=cv2.INTER_AREA) for p in (part0, part1)] + transforms, (cw, ch) = scale_transforms(cfg, [(p.shape[1], p.shape[0]) for p in halves]) + assert abs(cw - cfg.stitch_canvas[0] / 2) <= 1 + assert abs(ch - cfg.stitch_canvas[1] / 2) <= 1 + + half_cfg = StitchConfig( + stitch_enabled=True, + stitch_paths=cfg.stitch_paths, + stitch_transforms=tuple(tuple(float(v) for v in m.ravel()) for m in transforms), + stitch_canvas=(cw, ch), + stitch_sizes=tuple((p.shape[1], p.shape[0]) for p in halves), + ) + half = stitch_composite(halves, [None, None], half_cfg)[0] + ref = cv2.resize(full, (cw, ch), interpolation=cv2.INTER_AREA) + rms = float(np.sqrt(np.mean((half[20:-20, 20:-20] - ref[20:-20, 20:-20]) ** 2))) + assert rms < 0.03 + + +def test_stitch_hash_stable_and_order_sensitive(): + from negpy.services.assets.half_frame import base_hash + + h1 = stitch_hash(["aaa", "bbb"]) + assert h1 == stitch_hash(["aaa", "bbb"]) + assert h1 != stitch_hash(["bbb", "aaa"]) + assert h1.endswith("#stitch") + assert base_hash(h1) == h1.split("#", 1)[0] + + +def test_workspace_config_roundtrip_preserves_stitch(): + import json + from dataclasses import replace + + from negpy.domain.models import WorkspaceConfig + + stitch = StitchConfig( + stitch_enabled=True, + stitch_paths=("/p1.nef",), + stitch_transforms=((1.0, 0.0, 0.0, 0.0, 1.0, 0.0), (1.0, 0.0, 50.5, 0.0, 1.0, 2.5)), + stitch_canvas=(4000, 3000), + stitch_sizes=((3000, 2000), (3000, 2000)), + ) + cfg = replace(WorkspaceConfig(), stitch=stitch) + # JSON round-trip turns tuples into lists; from_flat_dict must coerce back + # or the frozen config loses hashability/equality. + data = json.loads(json.dumps(cfg.to_dict())) + restored = WorkspaceConfig.from_flat_dict(data) + assert restored.stitch == stitch + hash(restored.stitch) + + +def test_load_source_f32_stitches_composite(tmp_path): + """Decode funnel must assemble the composite (translation-only fixture: exact content).""" + import tifffile + from dataclasses import replace + + from negpy.domain.models import WorkspaceConfig + from negpy.services.rendering.image_processor import ImageProcessor + + rng = np.random.default_rng(3) + scene = (rng.random((80, 120, 3)) * 60000).astype(np.uint16) + p0 = tmp_path / "part0.tif" + p1 = tmp_path / "part1.tif" + tifffile.imwrite(str(p0), scene[:, :80]) + tifffile.imwrite(str(p1), scene[:, 40:]) + + stitch = StitchConfig( + stitch_enabled=True, + stitch_paths=(str(p1),), + stitch_transforms=((1.0, 0.0, 0.0, 0.0, 1.0, 0.0), (1.0, 0.0, 40.0, 0.0, 1.0, 0.0)), + stitch_canvas=(120, 80), + stitch_sizes=((80, 80), (80, 80)), + ) + params = replace(WorkspaceConfig(), stitch=stitch) + proc = ImageProcessor() + f32, ir, _ = proc._load_source_f32(str(p0), params) + assert ir is None + assert f32.shape == (80, 120, 3) + assert np.allclose(f32, scene.astype(np.float32) / 65535.0, atol=1e-3) + + again, _, _ = proc._load_source_f32(str(p0), params) + assert again is f32 # decode cache hit + moved = replace( + params, + stitch=replace(stitch, stitch_transforms=((1.0, 0.0, 0.0, 0.0, 1.0, 0.0), (1.0, 0.0, 41.0, 0.0, 1.0, 0.0))), + ) + fresh, _, _ = proc._load_source_f32(str(p0), moved) + assert fresh is not f32 # geometry change must invalidate the decode cache + + +def test_load_linear_preview_stitch(tmp_path): + """Preview path must assemble the composite and cache it under the stitch identity.""" + import tifffile + + from negpy.services.rendering.preview_manager import PreviewManager + + rng = np.random.default_rng(4) + scene = (rng.random((80, 120, 3)) * 60000).astype(np.uint16) + p0 = tmp_path / "part0.tif" + p1 = tmp_path / "part1.tif" + tifffile.imwrite(str(p0), scene[:, :80]) + tifffile.imwrite(str(p1), scene[:, 40:]) + stitch = StitchConfig( + stitch_enabled=True, + stitch_paths=(str(p1),), + stitch_transforms=((1.0, 0.0, 0.0, 0.0, 1.0, 0.0), (1.0, 0.0, 40.0, 0.0, 1.0, 0.0)), + stitch_canvas=(120, 80), + stitch_sizes=((80, 80), (80, 80)), + ) + pm = PreviewManager() + out, dims, meta = pm.load_linear_preview_stitch(str(p0), stitch, "Adobe RGB", use_camera_wb=False, file_hash="abc#stitch") + assert out.shape == (80, 120, 3) + assert dims == (80, 120) + assert meta.get("ir_preview") is None + assert np.allclose(out, scene.astype(np.float32) / 65535.0, atol=1e-3) + + again, _, _ = pm.load_linear_preview_stitch(str(p0), stitch, "Adobe RGB", use_camera_wb=False, file_hash="abc#stitch") + assert again is out # composite preview cache hit + + +def _worker(monkeypatch, buffers): + from negpy.desktop.workers.stitch import StitchWorker + + worker = StitchWorker() + monkeypatch.setattr( + worker._processor, + "_decode_oriented_f32", + lambda path, params, fast_decode=False: (buffers[path], None, "Adobe RGB"), + ) + return worker + + +def _worker_task(paths): + from dataclasses import replace as _r + + from negpy.desktop.workers.stitch import StitchTask + from negpy.domain.models import WorkspaceConfig + + files = tuple({"name": p.strip("/"), "path": p, "hash": f"h{i}"} for i, p in enumerate(paths)) + return StitchTask(files=files, params_by_path={p: _r(WorkspaceConfig()) for p in paths}) + + +def test_stitch_worker_emits_registered_payload(monkeypatch): + _, part0, part1, _ = _parts() + worker = _worker(monkeypatch, {"/a.nef": part0, "/b.nef": part1}) + results, errors = [], [] + worker.registered.connect(results.append) + worker.error.connect(errors.append) + worker.run(_worker_task(["/a.nef", "/b.nef"])) + assert not errors + payload = results[0] + assert [f["path"] for f in payload["files"]] == ["/a.nef", "/b.nef"] + assert len(payload["transforms"]) == 2 and len(payload["transforms"][0]) == 6 + assert payload["sizes"] == ((_P0_W, _H), (_P1_W, _H)) + assert abs(payload["canvas"][0] - _W) < 20 + + +def test_stitch_worker_error_on_unrelated(monkeypatch): + rng = np.random.default_rng(2) + a = np.clip(rng.random((300, 300, 3)), 0, 1).astype(np.float32) + b = np.clip(rng.random((300, 300, 3)), 0, 1).astype(np.float32) + worker = _worker(monkeypatch, {"/a.nef": a, "/b.nef": b}) + results, errors = [], [] + worker.registered.connect(results.append) + worker.error.connect(errors.append) + worker.run(_worker_task(["/a.nef", "/b.nef"])) + assert not results + assert errors and "overlap" in errors[0] + + +def test_stitch_worker_cancel_between_decodes(monkeypatch): + _, part0, part1, _ = _parts() + worker = _worker(monkeypatch, {"/a.nef": part0, "/b.nef": part1}) + + real_decode = worker._processor._decode_oriented_f32 + + def decode_then_cancel(path, params, fast_decode=False): + worker._cancel.set() + return real_decode(path, params, fast_decode) + + monkeypatch.setattr(worker._processor, "_decode_oriented_f32", decode_then_cancel) + results, cancels = [], [] + worker.registered.connect(results.append) + worker.cancelled.connect(lambda: cancels.append(True)) + worker.run(_worker_task(["/a.nef", "/b.nef"])) + assert cancels and not results + + +def test_attach_restored_stitches_rebuilds_asset(tmp_path): + """Session restore must rebuild a composite from the saved registration, not re-register.""" + from negpy.desktop.workers.render import AssetDiscoveryWorker + + p0 = tmp_path / "part0.nef" + p1 = tmp_path / "part1.nef" + for f in (p0, p1): + f.write_bytes(b"x") + assets = [{"name": "part0.nef", "path": str(p0), "hash": "h0"}] + stitches = { + str(p0): { + "paths": [str(p1)], + "transforms": [[1, 0, 0, 0, 1, 0], [1, 0, 40, 0, 1, 0]], + "canvas": [120, 80], + "sizes": [[80, 80], [80, 80]], + "hash": "digest#stitch", + } + } + out = AssetDiscoveryWorker()._attach_restored_stitches(assets, stitches) + assert out[0]["hash"] == "digest#stitch" + assert out[0]["stitch_paths"] == (str(p1),) + assert out[0]["name"].endswith("(Stitch)") + # A part gone missing on disk restores as a plain asset. + stitches[str(p0)]["paths"] = [str(tmp_path / "gone.nef")] + plain = AssetDiscoveryWorker()._attach_restored_stitches([dict(assets[0])], stitches) + assert "stitch_paths" not in plain[0] + + +def test_resolve_asset_stitch_injects_and_resets(): + from dataclasses import replace + + from negpy.desktop.session import resolve_asset_stitch + from negpy.domain.models import WorkspaceConfig + + asset = { + "path": "/p0", + "stitch_paths": ["/p1"], + "stitch_transforms": [[1, 0, 0, 0, 1, 0], [1, 0, 40, 0, 1, 0]], + "stitch_canvas": [120, 80], + "stitch_sizes": [[80, 80], [80, 80]], + } + out = resolve_asset_stitch(WorkspaceConfig(), asset) + assert out.stitch.stitch_enabled + assert out.stitch.stitch_paths == ("/p1",) + assert out.stitch.stitch_canvas == (120, 80) + hash(out.stitch) # lists from JSON/session must arrive as tuples + + leaked = replace(WorkspaceConfig(), stitch=out.stitch) + reset = resolve_asset_stitch(leaked, {"path": "/other"}) + assert reset.stitch == StitchConfig() + + +def test_apply_stitch_replaces_parts_with_composite(): + from unittest.mock import MagicMock + + from negpy.desktop.session import DesktopSessionManager + from negpy.infrastructure.storage.repository import StorageRepository + + repo = MagicMock(spec=StorageRepository) + repo.load_file_settings.return_value = None + repo.load_file_settings_by_path.return_value = None + repo.get_global_setting.side_effect = lambda key, default=None: default + repo.load_file_marks.return_value = {} + repo.get_max_history_index.return_value = 0 + session = DesktopSessionManager(repo) + session.state.uploaded_files = [ + {"name": "other.nef", "path": "/other", "hash": "hx"}, + {"name": "a.nef", "path": "/a", "hash": "ha"}, + {"name": "b.nef", "path": "/b", "hash": "hb"}, + ] + composite = { + "name": "a+b (Stitch)", + "path": "/a", + "hash": "digest#stitch", + "stitch_paths": ("/b",), + "stitch_transforms": ((1.0, 0.0, 0.0, 0.0, 1.0, 0.0), (1.0, 0.0, 40.0, 0.0, 1.0, 0.0)), + "stitch_canvas": (120, 80), + "stitch_sizes": ((80, 80), (80, 80)), + } + session.apply_stitch([1, 2], composite) + files = session.state.uploaded_files + assert [f["path"] for f in files] == ["/other", "/a"] + assert files[1]["hash"] == "digest#stitch" + assert session.state.selected_file_idx == 1 + saved = {c.args[0]: c.args[1] for c in repo.save_global_setting.call_args_list} + assert "/a" in saved.get("session_stitches", {}) + + +def test_stitch_real_panorama_samples(): + """End-to-end registration on the real camera-scan pair from discussion #555.""" + import os + + p0 = os.path.join("samples", "panorama", "Img427.nef") + p1 = os.path.join("samples", "panorama", "Img428.nef") + if not (os.path.exists(p0) and os.path.exists(p1)): + pytest.skip("panorama samples not present") + + from negpy.domain.models import WorkspaceConfig + from negpy.services.rendering.image_processor import ImageProcessor + + proc = ImageProcessor() + params = WorkspaceConfig() + parts = [proc._decode_oriented_f32(p, params, fast_decode=True)[0] for p in (p0, p1)] + + proxies = [build_proxy(p) for p in parts] + _, inliers = register_pair(proxies[0][0], proxies[1][0]) + assert inliers > 1000 + + transforms, (cw, ch) = register_parts(parts) + h, w = parts[0].shape[:2] + assert w <= cw < 2 * w + 20 + assert h <= ch < 2 * h + 20 + + cfg = StitchConfig( + stitch_enabled=True, + stitch_paths=(p1,), + stitch_transforms=tuple(tuple(float(v) for v in m.ravel()) for m in transforms), + stitch_canvas=(cw, ch), + stitch_sizes=tuple((p.shape[1], p.shape[0]) for p in parts), + ) + rgb, ir = stitch_composite(parts, [None, None], cfg) + assert rgb.shape == (ch, cw, 3) + assert ir is None + assert float(rgb.max()) <= 1.0 and float(rgb.min()) >= 0.0 + + +def test_stitch_token_identity(tmp_path): + assert stitch_token(StitchConfig()) == "" + p1 = tmp_path / "a.nef" + p1.write_bytes(b"x") + base = dict( + stitch_enabled=True, + stitch_paths=(str(p1),), + stitch_transforms=((1.0, 0.0, 0.0, 0.0, 1.0, 0.0), (1.0, 0.0, 50.0, 0.0, 1.0, 0.0)), + stitch_canvas=(100, 100), + stitch_sizes=((60, 100), (60, 100)), + ) + tok = stitch_token(StitchConfig(**base)) + assert tok.startswith("|stitch:") + moved = StitchConfig(**{**base, "stitch_transforms": ((1.0, 0.0, 0.0, 0.0, 1.0, 0.0), (1.0, 0.0, 60.0, 0.0, 1.0, 0.0))}) + assert stitch_token(moved) != tok + # Missing part file -> inactive token (same convention as rgbscan_token). + gone = StitchConfig(**{**base, "stitch_paths": (str(tmp_path / "missing.nef"),)}) + assert stitch_token(gone) == ""