feat: stitch frames captured as RGB-scan triplets

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**:

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)

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."""

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

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,

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(

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:

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)

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: