From a22b0fa1b7c5217c7f8f11aa73e590f3e8113a1f Mon Sep 17 00:00:00 2001 From: Marcin Zawalski Date: Sat, 1 Aug 2026 15:21:02 +0200 Subject: [PATCH] feat: stitch frames captured as RGB-scan triplets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stitching refused any RGB-scan frame. The reason was structural: StitchConfig carried one path per part, so the decode loop merged every part against the primary's green/blue exposures. StitchConfig now carries stitch_triplets — the (green, blue) pair per part, indexed like stitch_transforms/stitch_sizes with the primary at 0 — plus stitch_align. Each part is decoded against its own rgbscan config, which makes the existing per-part tail (triplet merge, flat-field, sensor-unmix skip) correct without further change. An empty tuple leaves saved composites alone. The triplet paths are folded into stitch_token, carried on the composite asset dict, persisted with the session and restored on launch; unstitch hands the parts back as triplet assets rather than loose exposures. Also fixes the RGB Scan / Half Frame toggles, which decomposed only green_path/blue_path and so dropped a composite's other parts from the session on re-discovery. Both now share one _component_paths helper. --- docs/PIPELINE.md | 1 + docs/USER_GUIDE.md | 6 + negpy/desktop/controller.py | 48 ++-- negpy/desktop/session.py | 4 + negpy/desktop/workers/render.py | 10 +- negpy/domain/models.py | 2 + negpy/features/stitch/models.py | 17 +- negpy/services/rendering/image_processor.py | 30 +- negpy/services/rendering/preview_manager.py | 8 +- tests/test_stitch.py | 292 ++++++++++++++++++++ 10 files changed, 392 insertions(+), 26 deletions(-) diff --git a/docs/PIPELINE.md b/docs/PIPELINE.md index 73f3c12d..258f8358 100644 --- a/docs/PIPELINE.md +++ b/docs/PIPELINE.md @@ -23,6 +23,7 @@ Here is what actually happens to your image. We apply these steps in order, pass * **Source corrections** (linear domain, before the log conversion): * **Flat-field** (`negpy.features.flatfield`): divides out illumination falloff using a blank reference frame. A per-channel gain map $\text{mean}(\text{blur})/\text{blur}$ (computed on a 256 px copy, clamped to $[0.25, 4]$) multiplies the linear source. The reference is decoded and the gain **baked once** into a profile (an `.npz` in `APP_CONFIG.flatfield_dir`, keyed by an opaque id), so the render never touches the original reference file — moving or deleting it is harmless. The per-image edit stores only the profile id; the render path resolves the gain through a provider (`set_gain_provider`, wired to `services/assets/flatfield.py` at startup) and caches it, and the profile id + a content token of the gain are folded into the render's source hash. * **Sensor crosstalk unmix** (`sensor_matrix`, `features/process/sensor.py`): for single-shot narrowband camera scans, the camera's CFA passbands overlap the light source's bands, so a pure R/G/B exposure leaks into the other channels. That is a fixed property of the sensor+light pair, independent of film. It is calibrated once from three bare-light exposures (response columns normalized to a unit diagonal so per-capture exposure cancels, then inverted) and applied as a 3×3 unmix of the **linear** capture, ahead of the log/inversion where the film-dye crosstalk below lives. The related **narrowband scan** toggle instead applies the bundled RGBScan *input* profile at the display/export boundary (an explicit Input ICC overrides it). + * **Composite assembly** (`features/rgbscan`, `features/stitch`): a frame may be built from several files before any of the above runs. An **RGB-scan triplet** takes each output channel from the exposure lit by that band (green/blue registered to red by phase correlation), so its channels never mixed in the sensor — which is why the crosstalk unmix above is skipped for triplets. A **stitch composite** decodes each overlapping part separately, warps them into a shared canvas by a registration fixed once at stitch time, matches per-channel gain across the overlap and feather-blends the seam. The two nest: every part carries its own triplet, and flat-field and unmix are applied **per part, never to the composite canvas** — a canvas-wide gain map would stretch across the seam. * **Log Conversion**: Film density is logarithmic ($D \propto \log E$). We convert the raw signal to log-space to align with the physics of the film layers: $$E_{log} = \log_{10}(I_{raw})$$ * **Bounding & Polarity**: diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index c2970046..26d8ecde 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -60,6 +60,12 @@ Below the toolbar: a **filter box** (substring match; toggle **`.*`** for regex) Right-clicking **empty space** in the film strip offers **Add files**, **Add folder** and **Clear all**, so those tools stay in reach part-way down a long roll instead of only at the top of the panel. Here **Clear all** always means the whole session, never just the selection. +#### Stitching a frame from several shots + +If one negative was captured in overlapping pieces (a copy stand at higher magnification than the frame), select the pieces and right-click → **Stitch selected frames**. NegPy finds the overlap, matches brightness across the seam and replaces the parts with a single wide composite named *a+b (Stitch)*. The parts' own edits stay on file, so right-click → **Unstitch** puts them back untouched. The registration is saved with the session and replayed on the next launch, so re-opening a composite costs nothing. + +This works on RGB-scan frames too: turn on **RGB Scan** first so each piece is already assembled from its own R/G/B triplet, then stitch the assembled frames. Each part keeps its own three exposures — nothing is shared between parts. + Narrow the panel and the toolbar buttons that no longer fit move into a **»** menu at its right edge, so the panel can be squeezed down to give the image more room without losing any tool. ### Triage (culling the roll) diff --git a/negpy/desktop/controller.py b/negpy/desktop/controller.py index df19503f..32a154b0 100644 --- a/negpy/desktop/controller.py +++ b/negpy/desktop/controller.py @@ -107,6 +107,19 @@ def _capture_import_key(path: str) -> str: return os.path.normcase(os.path.abspath(path)) +def _component_paths(files: List[Dict]) -> List[str]: + """Every source file behind the loaded assets, composites decomposed into their parts. + + Re-discovery over an asset list that only saw primaries would drop the rest.""" + paths: List[str] = [] + for f in files: + paths.append(f["path"]) + paths.extend(f[k] for k in ("green_path", "blue_path") if f.get(k)) + paths.extend(f.get("stitch_paths") or ()) + paths.extend(p for t in f.get("stitch_triplets") or () for p in t if p) + return list(dict.fromkeys(paths)) + + def _autocrop_fingerprint(config: WorkspaceConfig, workspace_color_space: str) -> tuple: """Identity of every setting that changes detection pixels or crop coordinates.""" geometry = config.geometry @@ -819,13 +832,7 @@ def set_rgb_scan_mode(self, enabled: bool) -> None: replace(self.state.config, process=replace(self.state.config.process, narrowband_scan=True)), persist=True ) self.request_render() - paths: List[str] = [] - for f in files: - paths.append(f["path"]) - for k in ("green_path", "blue_path"): - if f.get(k): - paths.append(f[k]) - self.request_asset_discovery(paths, replace_existing=True, reselect_path=self.state.current_file_path) + self.request_asset_discovery(_component_paths(files), replace_existing=True, reselect_path=self.state.current_file_path) def apply_scan_setup(self, capture: str, light: str) -> None: """Apply the scanning-setup wizard's answer: Linear RAW and Narrowband are rig @@ -891,13 +898,7 @@ def set_half_frame_mode(self, enabled: bool) -> None: files = self.session.state.uploaded_files if not files: return - paths: List[str] = [] - for f in files: - paths.append(f["path"]) - for k in ("green_path", "blue_path"): - if f.get(k): - paths.append(f[k]) - self.request_asset_discovery(paths, replace_existing=True, reselect_path=self.state.current_file_path) + self.request_asset_discovery(_component_paths(files), replace_existing=True, reselect_path=self.state.current_file_path) def _on_discovery_progress(self, current: int, total: int, name: str) -> None: self.set_status(f"HASHING {current}/{total}: {name}") @@ -1109,6 +1110,8 @@ def load_file(self, file_path: str, preserve_zoom: bool = False, force_detect: b stitch_transforms=stitch.stitch_transforms if stitch.stitch_enabled else (), stitch_canvas=stitch.stitch_canvas, stitch_sizes=stitch.stitch_sizes, + stitch_triplets=stitch.stitch_triplets if stitch.stitch_enabled else (), + stitch_align=stitch.stitch_align, flatfield_profile_id=flatfield.profile_id if (stitch.stitch_enabled and flatfield.apply) else "", ) ) @@ -2646,8 +2649,8 @@ def request_stitch_selected(self) -> None: 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) + if any(f.get("stitch_paths") for f in ordered): + self.set_status("Stitching an already-stitched frame is not supported", 4000) return if self._begin_batch("stitch", "Stitching frames", abortable=True) is None: return @@ -2662,6 +2665,7 @@ def _on_stitch_registered(self, payload: dict) -> None: self._end_batch("stitch") files = payload["files"] part_paths = [f["path"] for f in files] + triplets = tuple((f.get("green_path") or "", f.get("blue_path") or "") for f in files) composite = { "name": stitch_name(part_paths), "path": part_paths[0], @@ -2670,7 +2674,12 @@ def _on_stitch_registered(self, payload: dict) -> None: "stitch_transforms": payload["transforms"], "stitch_canvas": payload["canvas"], "stitch_sizes": payload["sizes"], + "stitch_triplets": triplets, + "stitch_align": bool(files[0].get("align", True)), } + if all(triplets[0]): + # Thumbnail decode and the sensor-unmix skip read the primary's pair from here. + composite.update(green_path=triplets[0][0], blue_path=triplets[0][1], align=composite["stitch_align"]) 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) @@ -2698,12 +2707,17 @@ def request_unstitch(self) -> None: if not parts: return paths = [asset["path"], *parts] + # Triplet parts must come back as triplet assets, not as loose exposures. + align = bool(asset.get("stitch_align", True)) + triplets = {path: [green, blue, align] for path, (green, blue) in zip(paths, asset.get("stitch_triplets") or ()) if green and blue} + for green, blue, _ in triplets.values(): + paths.extend((green, blue)) self.state.uploaded_files.pop(idx) self.session.state.thumbnails.pop(asset["name"], None) self.session.state.rendered_thumbnails.discard(asset["name"]) self.session.asset_model.refresh() self._pending_scanned_file = paths[0] - self.request_asset_discovery(paths) + self.request_asset_discovery(paths, restore_triplets=triplets or None) def _select_file_by_path(self, path: str) -> bool: """Find a file by path in uploaded_files and select it.""" diff --git a/negpy/desktop/session.py b/negpy/desktop/session.py index f7ef1fd8..7ec75d7c 100644 --- a/negpy/desktop/session.py +++ b/negpy/desktop/session.py @@ -368,6 +368,8 @@ def resolve_asset_stitch(params: WorkspaceConfig, asset: dict) -> WorkspaceConfi 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 ()), + stitch_triplets=tuple((str(t[0]), str(t[1])) for t in asset.get("stitch_triplets") or ()), + stitch_align=bool(asset.get("stitch_align", True)), ), ) return replace(params, stitch=StitchConfig()) @@ -1119,6 +1121,8 @@ def _persist_session(self) -> None: "transforms": [list(t) for t in f["stitch_transforms"]], "canvas": list(f["stitch_canvas"]), "sizes": [list(s) for s in f["stitch_sizes"]], + "triplets": [list(t) for t in f.get("stitch_triplets") or ()], + "align": bool(f.get("stitch_align", True)), "hash": f["hash"], } for f in self.state.uploaded_files diff --git a/negpy/desktop/workers/render.py b/negpy/desktop/workers/render.py index 29ed6043..92892699 100644 --- a/negpy/desktop/workers/render.py +++ b/negpy/desktop/workers/render.py @@ -165,6 +165,8 @@ class PreviewLoadTask: stitch_transforms: tuple[tuple[float, ...], ...] = () stitch_canvas: tuple[int, int] = (0, 0) stitch_sizes: tuple[tuple[int, int], ...] = () + stitch_triplets: tuple[tuple[str, str], ...] = () # per-part (green, blue) RGB-scan exposures + stitch_align: bool = True flatfield_profile_id: str = "" # per-part flat-field profile for stitch previews @@ -485,7 +487,9 @@ def _attach_restored_stitches(self, assets: list, stitches: dict) -> list: 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"]): + # Triplet exposures count as parts — one missing decodes that part red-only. + needed = [*(entry.get("paths") or ()), *(p for t in entry.get("triplets") or () for p in t if p)] if entry else [] + if entry and entry.get("paths") and all(os.path.exists(p) for p in needed): out.append( { **a, @@ -495,6 +499,8 @@ def _attach_restored_stitches(self, assets: list, stitches: dict) -> list: "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"]), + "stitch_triplets": tuple((str(t[0]), str(t[1])) for t in entry.get("triplets") or ()), + "stitch_align": bool(entry.get("align", True)), } ) else: @@ -577,6 +583,8 @@ def process(self, task: PreviewLoadTask) -> None: stitch_transforms=task.stitch_transforms, stitch_canvas=task.stitch_canvas, stitch_sizes=task.stitch_sizes, + stitch_triplets=task.stitch_triplets, + stitch_align=task.stitch_align, ) raw, dims, metadata = self._preview_service.load_linear_preview_stitch( task.file_path, diff --git a/negpy/domain/models.py b/negpy/domain/models.py index b7907e85..57567578 100644 --- a/negpy/domain/models.py +++ b/negpy/domain/models.py @@ -454,6 +454,8 @@ def _build_stitch(d: Dict[str, Any]) -> StitchConfig: 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", ())), + stitch_triplets=tuple((str(t[0]), str(t[1])) for t in d.get("stitch_triplets", ())), + stitch_align=bool(d.get("stitch_align", True)), ) return cls( diff --git a/negpy/features/stitch/models.py b/negpy/features/stitch/models.py index ec9fc28d..b129f2be 100644 --- a/negpy/features/stitch/models.py +++ b/negpy/features/stitch/models.py @@ -18,6 +18,13 @@ class StitchConfig: 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 + stitch_triplets: tuple[tuple[str, str], ...] = () # RGB-scan (green, blue) per part incl. primary + stitch_align: bool = True # sub-pixel registration within each part's triplet + + +def stitch_has_triplets(config: StitchConfig) -> bool: + """True when any part is assembled from an R/G/B exposure triplet.""" + return any(green and blue for green, blue in config.stitch_triplets) def stitch_token(config: StitchConfig) -> str: @@ -30,9 +37,17 @@ def stitch_token(config: StitchConfig) -> str: parts.append(f"{path}:{os.path.getmtime(path)}") except OSError: return "" + for green, blue in config.stitch_triplets: + for path in (green, blue): + if not path: + continue + 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}" + return f"|stitch:{':'.join(parts)}:{digest}:a{int(config.stitch_align)}" def stitch_name(part_paths: Sequence[str]) -> str: diff --git a/negpy/services/rendering/image_processor.py b/negpy/services/rendering/image_processor.py index 89d64e07..e4547170 100644 --- a/negpy/services/rendering/image_processor.py +++ b/negpy/services/rendering/image_processor.py @@ -38,9 +38,9 @@ route_ir_defects, ) from negpy.features.rgbscan.logic import merge_rgb_triplet, rgbscan_token -from negpy.features.rgbscan.models import is_rgb_triplet +from negpy.features.rgbscan.models import RgbScanConfig, is_rgb_triplet from negpy.features.stitch.logic import stitch_composite -from negpy.features.stitch.models import stitch_token +from negpy.features.stitch.models import stitch_has_triplets, stitch_token from negpy.domain.interfaces import PipelineContext from negpy.services.rendering.engine import DarkroomEngine from negpy.services.rendering.gpu_engine import GPUEngine @@ -101,6 +101,23 @@ def _detection_downsample(buf: np.ndarray) -> np.ndarray: return cv2.resize(buf, (max(1, int(round(w * s))), max(1, int(round(h * s)))), interpolation=cv2.INTER_AREA) +def _part_params(params: WorkspaceConfig, index: int) -> WorkspaceConfig: + """Params for stitch part ``index``, carrying that part's own R/G/B exposures. + + Empty ``stitch_triplets`` (composites registered before triplet support) falls back + to ``params.rgbscan``.""" + triplets = params.stitch.stitch_triplets + if index >= len(triplets): + return params + green, blue = triplets[index] + rgbscan = ( + RgbScanConfig(enabled=True, green_path=green, blue_path=blue, align=params.stitch.stitch_align) + if (green and blue) + else RgbScanConfig() + ) + return dc_replace(params, rgbscan=rgbscan) + + class ImageProcessor: """ Coordinates multi-backend image processing. @@ -309,7 +326,7 @@ def run_pipeline( # come from _load_source_f32, which already applied it; triplet composites take # each channel from its own single-band exposure, so unmixing them would inject # crosstalk that was never captured. - if not skip_flatfield and not is_rgb_triplet(settings.rgbscan): + if not skip_flatfield and not is_rgb_triplet(settings.rgbscan) and not stitch_has_triplets(settings.stitch): img = apply_sensor_correction(img, effective_sensor_matrix(settings.process)) h_orig, w_cols = img.shape[:2] # Fold the buffer resolution into source_hash: toggling HQ re-decodes the same @@ -456,11 +473,12 @@ def _load_source_f32( """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``. + registration stored in ``params.stitch``, each part against its own rgbscan + config rather than the primary's. Returns (f32_buffer, ir_buffer, source_color_space). """ - is_triplet = is_rgb_triplet(params.rgbscan) + is_triplet = is_rgb_triplet(params.rgbscan) or stitch_has_triplets(params.stitch) # Narrowband triplet channels don't survive half_size CFA binning. fast_decode = fast_decode and not is_triplet @@ -485,7 +503,7 @@ def _load_source_f32( 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) + f32, ir, cs = self._decode_oriented_f32(path, _part_params(params, i), fast_decode) if i == 0: source_cs = cs parts.append(f32) diff --git a/negpy/services/rendering/preview_manager.py b/negpy/services/rendering/preview_manager.py index c1904478..21c8dbda 100644 --- a/negpy/services/rendering/preview_manager.py +++ b/negpy/services/rendering/preview_manager.py @@ -427,8 +427,14 @@ def load_linear_preview_stitch( parts, irs = [], [] meta: dict = {} for i, path in enumerate((primary_path, *stitch.stitch_paths)): + green, blue = stitch.stitch_triplets[i] if i < len(stitch.stitch_triplets) else ("", "") # 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) + if green and blue: + out, _, part_meta = self.load_linear_preview_rgb( + path, green, blue, color_space, use_camera_wb, full_resolution, None, align=stitch.stitch_align + ) + else: + 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: diff --git a/tests/test_stitch.py b/tests/test_stitch.py index ac145edb..84b177c1 100644 --- a/tests/test_stitch.py +++ b/tests/test_stitch.py @@ -478,3 +478,295 @@ def test_stitch_token_identity(tmp_path): # Missing part file -> inactive token (same convention as rgbscan_token). gone = StitchConfig(**{**base, "stitch_paths": (str(tmp_path / "missing.nef"),)}) assert stitch_token(gone) == "" + + +# ── RGB-scan triplet parts ──────────────────────────────────────────── + + +def _triplet_tifs(tmp_path, tag, values, width=80, offset=0): + """Three constant-colour exposures; each carries its own channel at ``values``.""" + import tifffile + + paths = [] + for i, name in enumerate("rgb"): + frame = np.zeros((80, width, 3), np.uint16) + frame[..., i] = values[i] + frame[..., (i + 1) % 3] = offset + i # decoy: must not reach the composite + p = tmp_path / f"{tag}_{name}.tif" + tifffile.imwrite(str(p), frame) + paths.append(str(p)) + return paths + + +def _side_by_side_triplet_config(part0, part1): + """Two 80px parts laid out edge to edge — under _MIN_OVERLAP_PX, so no gain + compensation smears one part's values into the other's.""" + return StitchConfig( + stitch_enabled=True, + stitch_paths=(part1[0],), + stitch_transforms=((1.0, 0.0, 0.0, 0.0, 1.0, 0.0), (1.0, 0.0, 80.0, 0.0, 1.0, 0.0)), + stitch_canvas=(160, 80), + stitch_sizes=((80, 80), (80, 80)), + stitch_triplets=((part0[1], part0[2]), (part1[1], part1[2])), + stitch_align=False, + ) + + +def test_stitch_token_covers_triplets(tmp_path): + """Swapping a part's green/blue exposure must invalidate every render cache.""" + part = tmp_path / "a.nef" + green = tmp_path / "a_g.nef" + blue = tmp_path / "a_b.nef" + for f in (part, green, blue): + f.write_bytes(b"x") + base = dict( + stitch_enabled=True, + stitch_paths=(str(part),), + 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)), + ) + plain = stitch_token(StitchConfig(**base)) + with_triplet = stitch_token(StitchConfig(**base, stitch_triplets=(("", ""), (str(green), str(blue))))) + assert with_triplet != plain + unaligned = stitch_token(StitchConfig(**base, stitch_triplets=(("", ""), (str(green), str(blue))), stitch_align=False)) + assert unaligned != with_triplet + # A vanished exposure is as fatal as a vanished part -> inactive token. + gone = stitch_token(StitchConfig(**base, stitch_triplets=(("", ""), (str(tmp_path / "no.nef"), str(blue))))) + assert gone == "" + + +def test_workspace_config_roundtrip_preserves_stitch_triplets(): + from dataclasses import replace + + from negpy.domain.models import WorkspaceConfig + + stitch = StitchConfig( + stitch_enabled=True, + stitch_paths=("/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)), + stitch_triplets=(("/p0g", "/p0b"), ("/p1g", "/p1b")), + stitch_align=False, + ) + restored = WorkspaceConfig.from_flat_dict(replace(WorkspaceConfig(), stitch=stitch).to_dict()) + assert restored.stitch == stitch + hash(restored.stitch) + + +def test_load_source_f32_merges_each_part_with_its_own_triplet(tmp_path): + """Each part merges with its own green/blue exposures, never the primary's.""" + from dataclasses import replace + + from negpy.domain.models import WorkspaceConfig + from negpy.features.rgbscan.models import RgbScanConfig + from negpy.services.rendering.image_processor import ImageProcessor + + part0 = _triplet_tifs(tmp_path, "p0", (1000, 2000, 3000)) + part1 = _triplet_tifs(tmp_path, "p1", (1100, 2100, 3100), offset=7) + stitch = _side_by_side_triplet_config(part0, part1) + params = replace( + WorkspaceConfig(), + stitch=stitch, + rgbscan=RgbScanConfig(enabled=True, green_path=part0[1], blue_path=part0[2], align=False), + ) + + f32, _, _ = ImageProcessor()._load_source_f32(part0[0], params) + assert f32.shape == (80, 160, 3) + left = f32[:, 5:75].reshape(-1, 3).mean(axis=0) * 65535.0 + right = f32[:, 85:155].reshape(-1, 3).mean(axis=0) * 65535.0 + assert np.allclose(left, (1000, 2000, 3000), atol=2) + assert np.allclose(right, (1100, 2100, 3100), atol=2) + + +def test_load_linear_preview_stitch_merges_triplet_parts(tmp_path): + """Preview path must route each triplet part through the RGB merge too.""" + from negpy.services.rendering.preview_manager import PreviewManager + + part0 = _triplet_tifs(tmp_path, "p0", (1000, 2000, 3000)) + part1 = _triplet_tifs(tmp_path, "p1", (1100, 2100, 3100), offset=7) + stitch = _side_by_side_triplet_config(part0, part1) + + out, dims, _ = PreviewManager().load_linear_preview_stitch(part0[0], stitch, "Adobe RGB", use_camera_wb=False) + assert dims == (80, 160) + arr = np.asarray(out) + assert np.allclose(arr[:, 5:75].reshape(-1, 3).mean(axis=0) * 65535.0, (1000, 2000, 3000), atol=2) + assert np.allclose(arr[:, 85:155].reshape(-1, 3).mean(axis=0) * 65535.0, (1100, 2100, 3100), atol=2) + + +def test_resolve_asset_stitch_carries_triplets(): + 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]], + "stitch_triplets": [["/p0g", "/p0b"], ["/p1g", "/p1b"]], + "stitch_align": False, + } + out = resolve_asset_stitch(WorkspaceConfig(), asset) + assert out.stitch.stitch_triplets == (("/p0g", "/p0b"), ("/p1g", "/p1b")) + assert out.stitch.stitch_align is False + hash(out.stitch) + + +def test_attach_restored_stitches_needs_the_triplet_exposures(tmp_path): + from negpy.desktop.workers.render import AssetDiscoveryWorker + + files = {n: tmp_path / n for n in ("p0.nef", "p1.nef", "p1_g.nef", "p1_b.nef")} + for f in files.values(): + f.write_bytes(b"x") + assets = [{"name": "p0.nef", "path": str(files["p0.nef"]), "hash": "h0"}] + entry = { + "paths": [str(files["p1.nef"])], + "transforms": [[1, 0, 0, 0, 1, 0], [1, 0, 40, 0, 1, 0]], + "canvas": [120, 80], + "sizes": [[80, 80], [80, 80]], + "triplets": [["", ""], [str(files["p1_g.nef"]), str(files["p1_b.nef"])]], + "align": False, + "hash": "digest#stitch", + } + out = AssetDiscoveryWorker()._attach_restored_stitches([dict(assets[0])], {str(files["p0.nef"]): entry}) + assert out[0]["stitch_triplets"] == (("", ""), (str(files["p1_g.nef"]), str(files["p1_b.nef"]))) + assert out[0]["stitch_align"] is False + + entry["triplets"][1][0] = str(tmp_path / "gone.nef") + plain = AssetDiscoveryWorker()._attach_restored_stitches([dict(assets[0])], {str(files["p0.nef"]): entry}) + assert "stitch_paths" not in plain[0] + + +def _mock_controller(files, selected): + from unittest.mock import MagicMock + + from negpy.desktop.session import AppState + + c = MagicMock() + c.state = AppState() + c.session.state = c.state + c.state.uploaded_files = files + c.state.selected_indices = selected + c.state.selected_file_idx = selected[0] if selected else -1 + c._batch_busy.return_value = False + c._begin_batch.return_value = 1 + return c + + +def _triplet_asset(tag, align=True): + return { + "name": f"{tag} (RGB)", + "path": f"/{tag}_r.raf", + "hash": f"h_{tag}", + "green_path": f"/{tag}_g.raf", + "blue_path": f"/{tag}_b.raf", + "align": align, + } + + +def test_request_stitch_selected_accepts_triplets(): + from negpy.desktop.controller import AppController + + c = _mock_controller([_triplet_asset("a"), _triplet_asset("b")], [0, 1]) + AppController.request_stitch_selected(c) + task = c.stitch_requested.emit.call_args.args[0] + assert [f["path"] for f in task.files] == ["/a_r.raf", "/b_r.raf"] + + +def test_on_stitch_registered_stores_per_part_triplets(): + from negpy.desktop.controller import AppController + + c = _mock_controller([_triplet_asset("a"), _triplet_asset("b")], [0, 1]) + payload = { + "files": [_triplet_asset("a"), _triplet_asset("b")], + "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)), + "canvas": (120, 80), + "sizes": ((80, 80), (80, 80)), + } + AppController._on_stitch_registered(c, payload) + composite = c.session.apply_stitch.call_args.args[1] + assert composite["stitch_triplets"] == (("/a_g.raf", "/a_b.raf"), ("/b_g.raf", "/b_b.raf")) + assert composite["stitch_align"] is True + assert composite["green_path"] == "/a_g.raf" and composite["blue_path"] == "/a_b.raf" + + +def test_request_unstitch_restores_triplet_assets(): + from negpy.desktop.controller import AppController + + composite = { + "name": "a+b (Stitch)", + "path": "/a_r.raf", + "hash": "digest#stitch", + "stitch_paths": ("/b_r.raf",), + "stitch_triplets": (("/a_g.raf", "/a_b.raf"), ("/b_g.raf", "/b_b.raf")), + "stitch_align": True, + } + c = _mock_controller([composite], [0]) + AppController.request_unstitch(c) + paths, kwargs = c.request_asset_discovery.call_args.args[0], c.request_asset_discovery.call_args.kwargs + assert set(paths) == {"/a_r.raf", "/b_r.raf", "/a_g.raf", "/a_b.raf", "/b_g.raf", "/b_b.raf"} + assert kwargs["restore_triplets"] == { + "/a_r.raf": ["/a_g.raf", "/a_b.raf", True], + "/b_r.raf": ["/b_g.raf", "/b_b.raf", True], + } + + +def test_component_paths_decomposes_composites(): + from negpy.desktop.controller import _component_paths + + composite = { + "path": "/a_r.raf", + "green_path": "/a_g.raf", + "blue_path": "/a_b.raf", + "stitch_paths": ("/b_r.raf",), + "stitch_triplets": (("/a_g.raf", "/a_b.raf"), ("/b_g.raf", "/b_b.raf")), + } + assert set(_component_paths([composite])) == { + "/a_r.raf", + "/a_g.raf", + "/a_b.raf", + "/b_r.raf", + "/b_g.raf", + "/b_b.raf", + } + + +def test_stitch_real_rgb_triplet_samples(): + """End-to-end on the real narrowband capture: two overlapping parts, three + exposures each (samples/stitch/Roll001_Frame00{1,2}_{R,G,B}.raf).""" + import os + + from negpy.services.rendering.preview_manager import PreviewManager + + def triplet(frame): + return [os.path.join("samples", "stitch", f"Roll001_Frame{frame}_{ch}.raf") for ch in "RGB"] + + part0, part1 = triplet("001"), triplet("002") + if not all(os.path.exists(p) for p in (*part0, *part1)): + pytest.skip("RGB-triplet stitch samples not present") + + pm = PreviewManager() + merged = [ + np.asarray(pm.load_linear_preview_rgb(r, g, b, "Adobe RGB", use_camera_wb=False)[0], dtype=np.float32) for r, g, b in (part0, part1) + ] + transforms, (cw, ch) = register_parts(merged) + h, w = merged[0].shape[:2] + assert w < cw < 2 * w # the parts overlap, so the canvas is wider than one and narrower than two + + cfg = StitchConfig( + stitch_enabled=True, + stitch_paths=(part1[0],), + 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 merged), + stitch_triplets=((part0[1], part0[2]), (part1[1], part1[2])), + ) + out, dims, _ = pm.load_linear_preview_stitch(part0[0], cfg, "Adobe RGB", use_camera_wb=False) + arr = np.asarray(out, dtype=np.float32) + assert arr.shape == (ch, cw, 3) + assert 0.0 <= float(arr.min()) and float(arr.max()) <= 1.0 + # A red-only merge collapses the three channels onto each other. + means = [float(arr[..., i].mean()) for i in range(3)] + assert max(means) - min(means) > 0.01