refactor(flatfield): one helper for the render's flat-fielded source

negpy/desktop/controller.py

@@ -128,7 +128,6 @@
     remap_workspace_config,
     remember_split_scans,
     saved_crop_rect,
-    slice_chain,
     split_scans,
 )
 from negpy.services.export.templating import path_safe, render_export_filename
@@ -160,7 +159,6 @@
 )
 from negpy.features.altprocess.models import AltProcess
 from negpy.features.finish.models import FinishConfig
-from negpy.features.flatfield.logic import apply_flatfield
 from negpy.features.geometry.logic import (
     apply_fine_rotation,
     autocrop_detection_key,
@@ -212,7 +210,7 @@
 from negpy.services.rendering.prefetch_policy import MIN_RAM_RESERVE_BYTES
 from negpy.services.rendering.preview_manager import PreviewManager
 from negpy.services.rendering.source_identity import source_token
-from negpy.services.rendering.lens import lens_decode_token, metadata_lens_corrections
+from negpy.services.rendering.lens import flatfield_source, lens_decode_token, metadata_lens_corrections
 from negpy.services.view.coordinate_mapping import CoordinateMapping
 
 logger = get_logger(__name__)
@@ -3664,16 +3662,7 @@ def auto_skew_frame(self) -> None:
         config = self.state.config
         geo = config.geometry
         # Every detection path reads the flat-fielded source.
-        if config.stitch.stitch_enabled or metadata_lens_corrections(config):
-            source = raw
-        else:
-            cuts = self._preview_gain_slices(self.active_diptych())
-            source = apply_flatfield(
-                raw,
-                config.flatfield,
-                (lambda g: slice_chain(g, cuts)) if cuts else None,
-                orientation=self.state.preview_orientation,
-            )
+        source = flatfield_source(raw, config, self._preview_gain_slices(self.active_diptych()), orientation=self.state.preview_orientation)
         QApplication.setOverrideCursor(Qt.CursorShape.WaitCursor)
         try:
             fit = squared_geometry(source, geo, config.process.process_mode)

negpy/desktop/workers/render.py

@@ -20,10 +20,9 @@
     waveform,
     waveform_color,
 )
-from negpy.features.flatfield.logic import apply_flatfield
 from negpy.features.flatfield.models import FlatFieldConfig
 from negpy.features.lens.models import LensCorrections
-from negpy.services.rendering.lens import lens_decode_token, metadata_lens_corrections
+from negpy.services.rendering.lens import flatfield_source, lens_decode_token, metadata_lens_corrections
 from negpy.features.hdr.models import HdrConfig, hdr_active
 from negpy.features.geometry.batch_autocrop import CropEvidence, detect_crop_candidate, resolve_roll_crops
 from negpy.features.process.capture_color import wb_only_cam_xyz
@@ -1627,18 +1626,7 @@ def _frame_evidence(
                 return None, None
 
             config = frame.config
-            if config.stitch.stitch_enabled or metadata_lens_corrections(config):
-                corrected = raw
-            else:
-                from negpy.services.assets.half_frame import slice_chain
-
-                cuts = gain_slices_for_asset(file_info)
-                corrected = apply_flatfield(
-                    raw,
-                    config.flatfield,
-                    (lambda g: slice_chain(g, cuts)) if cuts else None,
-                    orientation=meta.get("orientation", 1),
-                )
+            corrected = flatfield_source(raw, config, gain_slices_for_asset(file_info), orientation=meta.get("orientation", 1))
             geometry = config.geometry
             keystone = None
             if squares_on_arm(geometry):
@@ -1996,19 +1984,14 @@ async def _analyze_file(frame: NormalizationInput):
                         params,
                         task.workspace_color_space,
                     )
-                    # The flat field the render applies, behind the same gate run_pipeline uses.
-                    if not params.stitch.stitch_enabled and not metadata_lens_corrections(params):
-                        from negpy.features.flatfield.logic import apply_flatfield
-                        from negpy.services.assets.half_frame import slice_chain
-
-                        cuts = gain_slices_for_asset(f_info)
-                        raw = await asyncio.to_thread(
-                            apply_flatfield,
-                            raw,
-                            params.flatfield,
-                            (lambda g: slice_chain(g, cuts)) if cuts else None,
-                            orientation=decode_meta.get("orientation", 1),
-                        )
+                    # The flat field the render applies.
+                    raw = await asyncio.to_thread(
+                        flatfield_source,
+                        raw,
+                        params,
+                        gain_slices_for_asset(f_info),
+                        orientation=decode_meta.get("orientation", 1),
+                    )
                     # Bounds must be measured on the same channel mix the preview normalizes.
                     sensor_matrix = effective_sensor_matrix(params.process)
                     if sensor_matrix is not None and preview_takes_unmix(params):

negpy/services/rendering/image_processor.py

@@ -40,7 +40,7 @@
 from negpy.features.exposure.models import RenderIntent
 from negpy.features.flatfield.logic import apply_flatfield, flatfield_token
 from negpy.features.lab.logic import apply_resize_sharpening
-from negpy.services.rendering.lens import lens_decode_token, metadata_lens_corrections, prepare_lens_source
+from negpy.services.rendering.lens import flatfield_source, lens_decode_token, metadata_lens_corrections, prepare_lens_source
 from negpy.features.geometry.logic import autocrop_detection_key, resolve_autocrop_rect
 from negpy.features.geometry.models import squares_on_arm
 from negpy.features.geometry.skew import squared_geometry
@@ -752,15 +752,8 @@ def run_pipeline(
             img = self._precorrect_value
         else:
             source = img
-            if not skip_flatfield and not settings.stitch.stitch_enabled and not metadata_lens_corrections(settings):
-                from negpy.services.assets.half_frame import slice_chain
-
-                img = apply_flatfield(
-                    img,
-                    settings.flatfield,
-                    (lambda g: slice_chain(g, gain_slices)) if gain_slices else None,
-                    orientation=source_orientation,
-                )
+            if not skip_flatfield:
+                img = flatfield_source(img, settings, gain_slices, orientation=source_orientation)
             # Sensor unmix is a source pre-correction like flat-field. skip_flatfield buffers
             # come from _load_source_f32, which already applied it.
             if not skip_flatfield and preview_takes_unmix(settings):

negpy/services/rendering/lens.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 from negpy.domain.types import ImageBuffer
 from negpy.features.flatfield.logic import apply_flatfield, flatfield_token
@@ -8,6 +8,7 @@
 from negpy.features.hdr.models import hdr_active
 from negpy.features.rgbscan.models import is_rgb_triplet
 from negpy.features.stitch.models import stitch_active
+from negpy.services.assets.half_frame import slice_chain
 
 if TYPE_CHECKING:
     from negpy.domain.models import WorkspaceConfig
@@ -20,6 +21,22 @@ def metadata_lens_corrections(config: "WorkspaceConfig") -> LensCorrections:
     return LensCorrections(config.geometry.lens_distortion_from_metadata, config.geometry.lens_ca_from_metadata)
 
 
+def flatfield_source(
+    img: ImageBuffer,
+    config: "WorkspaceConfig",
+    gain_slices: tuple = (),
+    *,
+    orientation: Optional[int],
+) -> ImageBuffer:
+    """*img* flat-fielded as the render's source is. ``gain_slices`` are the ``slice_half`` cuts
+    that took *img* out of the decoded frame. A stitch decode corrects its parts, and the
+    embedded-lens decode applies the flat field before its warp."""
+    if config.stitch.stitch_enabled or metadata_lens_corrections(config):
+        return img
+    gain_crop = (lambda g: slice_chain(g, gain_slices)) if gain_slices else None
+    return apply_flatfield(img, config.flatfield, gain_crop, orientation=orientation)
+
+
 def lens_decode_token(corrections: LensCorrections, flatfield: FlatFieldConfig) -> str:
     return f"|embedded-lens-v3-d{int(corrections.distortion)}-ca{int(corrections.ca)}" + flatfield_token(flatfield) if corrections else ""
 

tests/test_batch_autocrop_worker.py

@@ -5,6 +5,7 @@
 from negpy.features.lens.models import LensCorrections
 
 import negpy.desktop.workers.render as render_workers
+import negpy.services.rendering.lens as lens_module
 from negpy.desktop.workers.render import (
     BatchAutoCropInput,
     BatchAutoCropResult,
@@ -181,7 +182,7 @@ def _detect(key, transformed, *, target_ratio, rebate_trim=1.0):
         captured["rebate_trim"] = rebate_trim
         return SimpleNamespace(key=key)
 
-    monkeypatch.setattr(render_workers, "apply_flatfield", _apply_flatfield)
+    monkeypatch.setattr(lens_module, "apply_flatfield", _apply_flatfield)
     monkeypatch.setattr(render_workers, "GeometryProcessor", _GeometryProcessor)
     monkeypatch.setattr(render_workers, "detect_crop_candidate", _detect)
     monkeypatch.setattr(render_workers, "resolve_roll_crops", lambda _evidence: [])
@@ -214,7 +215,7 @@ def _flatfield_config() -> WorkspaceConfig:
 def test_batch_autocrop_cuts_the_flatfield_gain_to_the_half(qapp, monkeypatch) -> None:
     _stub_detection(monkeypatch)
     crops: list = []
-    monkeypatch.setattr(render_workers, "apply_flatfield", lambda image, _config, gain_crop=None, **_kw: crops.append(gain_crop) or image)
+    monkeypatch.setattr(lens_module, "apply_flatfield", lambda image, _config, gain_crop=None, **_kw: crops.append(gain_crop) or image)
     frame = _input("frame", _flatfield_config())
     frame.file_info.update(half=2, split_x=0.5, gutter_thickness=0.0)
 
@@ -227,7 +228,7 @@ def test_batch_autocrop_cuts_the_flatfield_gain_to_the_half(qapp, monkeypatch) -
 def test_batch_autocrop_skips_the_flatfield_on_a_stitch(qapp, monkeypatch) -> None:
     _stub_detection(monkeypatch)
     calls: list = []
-    monkeypatch.setattr(render_workers, "apply_flatfield", lambda image, *_args, **_kwargs: calls.append(1) or image)
+    monkeypatch.setattr(lens_module, "apply_flatfield", lambda image, *_args, **_kwargs: calls.append(1) or image)
     config = _flatfield_config()
     config = replace(config, stitch=replace(config.stitch, stitch_enabled=True))
 

tests/test_controller.py

@@ -210,7 +210,7 @@ def test_auto_skew_measures_the_flat_fielded_source(self):
         measured = FrameSkew(fine_rotation=-1.0, converge_v=None, converge_h=None, confidence=0.9, lines=(), residual=0.0)
 
         with (
-            patch("negpy.desktop.controller.apply_flatfield", return_value=flattened) as bake,
+            patch("negpy.services.rendering.lens.apply_flatfield", return_value=flattened) as bake,
             patch("negpy.features.geometry.skew.trusted_frame_skew", return_value=measured) as fit,
         ):
             self.controller.auto_skew_frame()
@@ -227,7 +227,7 @@ def test_auto_skew_cuts_the_flatfield_gain_to_the_active_half(self):
         with (
             patch.object(self.controller, "active_diptych", return_value=None),
             patch.object(self.controller, "_active_half", return_value=half),
-            patch("negpy.desktop.controller.apply_flatfield", side_effect=lambda image, *_, **__: image) as bake,
+            patch("negpy.services.rendering.lens.apply_flatfield", side_effect=lambda image, *_, **__: image) as bake,
             patch("negpy.features.geometry.skew.trusted_frame_skew", return_value=None),
         ):
             self.controller.auto_skew_frame()
@@ -243,7 +243,7 @@ def test_auto_skew_cuts_the_flatfield_gain_to_the_diptych_crop(self):
 
         with (
             patch.object(self.controller, "active_diptych", return_value=dip),
-            patch("negpy.desktop.controller.apply_flatfield", side_effect=lambda image, *_, **__: image) as bake,
+            patch("negpy.services.rendering.lens.apply_flatfield", side_effect=lambda image, *_, **__: image) as bake,
             patch("negpy.features.geometry.skew.trusted_frame_skew", return_value=None),
         ):
             self.controller.auto_skew_frame()
@@ -259,7 +259,7 @@ def test_auto_skew_skips_the_flatfield_on_a_stitch(self):
         self.controller.set_status = MagicMock()
 
         with (
-            patch("negpy.desktop.controller.apply_flatfield") as bake,
+            patch("negpy.services.rendering.lens.apply_flatfield") as bake,
             patch("negpy.features.geometry.skew.trusted_frame_skew", return_value=None),
         ):
             self.controller.auto_skew_frame()

tests/test_image_processor.py

@@ -414,13 +414,13 @@ def test_repair_falls_back_to_a_whole_frame_pass_when_defects_swarm() -> None:
 def test_run_pipeline_skip_flatfield(monkeypatch) -> None:
     """The export/contact-sheet CPU fallback passes an already-flat-fielded buffer;
     skip_flatfield must prevent a second apply_flatfield (the latent double-apply)."""
-    import negpy.services.rendering.image_processor as ip
+    import negpy.services.rendering.lens as lens
 
     service = ImageProcessor()
     monkeypatch.setattr(service.engine_cpu, "process", lambda img, s, sh, ctx: img)
     calls = []
-    real = ip.apply_flatfield
-    monkeypatch.setattr(ip, "apply_flatfield", lambda img, ff, *a, **kw: (calls.append(1), real(img, ff, *a, **kw))[1])
+    real = lens.apply_flatfield
+    monkeypatch.setattr(lens, "apply_flatfield", lambda img, ff, *a, **kw: (calls.append(1), real(img, ff, *a, **kw))[1])
 
     img = np.full((64, 64, 3), 0.5, dtype=np.float32)
     service.run_pipeline(img, WorkspaceConfig(), "h", render_size_ref=512, prefer_gpu=False, readback_metrics=False)

tests/test_metadata_lens.py

@@ -18,7 +18,7 @@
 from negpy.features.lens.warps import IDENTITY, RectilinearWarp, SonyWarp
 from negpy.infrastructure.loaders.lens_metadata import bind_decode, parse_opcodes, read_lens_metadata
 from negpy.kernel.image.logic import apply_exif_orientation
-from negpy.services.rendering.lens import lens_decode_token, metadata_lens_corrections, prepare_lens_source
+from negpy.services.rendering.lens import flatfield_source, lens_decode_token, metadata_lens_corrections, prepare_lens_source
 from negpy.services.rendering.preview_cache import PreviewCacheKey
 from negpy.services.rendering.source_identity import source_token
 
@@ -317,6 +317,31 @@ def test_bind_decode_checks_coordinate_compatibility():
     assert not bind_decode(lens, cropped_fallback, fallback=True).available
 
 
+def test_flatfield_source_gates_cuts_and_turns_the_gain(monkeypatch):
+    from negpy.services.rendering import lens as service
+
+    calls = []
+    monkeypatch.setattr(
+        service, "apply_flatfield", lambda img, config, gain_crop=None, *, orientation: calls.append((gain_crop, orientation)) or img + 1
+    )
+    image = np.zeros((40, 30, 3), np.float32)
+    config = WorkspaceConfig()
+
+    np.testing.assert_array_equal(flatfield_source(image, config, orientation=6), 1.0)
+    assert calls == [(None, 6)]
+    calls.clear()
+
+    flatfield_source(image, config, ((2, 0.5, None, 0.0, "x"),), orientation=1)
+    assert calls[0][0](np.zeros((40, 60, 3), np.float32)).shape == (40, 30, 3)
+    calls.clear()
+
+    stitched = replace(config, stitch=replace(config.stitch, stitch_enabled=True))
+    lens = replace(config, geometry=replace(config.geometry, lens_distortion_from_metadata=True))
+    assert flatfield_source(image, stitched, orientation=1) is image
+    assert flatfield_source(image, lens, orientation=1) is image
+    assert calls == []
+
+
 def test_flatfield_is_applied_before_the_lens_warp(monkeypatch):
     from negpy.services.rendering import lens as service