Reduce memory use for Linear DNG decoding and CPU export

CLAUDE.md

@@ -47,7 +47,7 @@ Migrations that rewrite *rows* rather than a config payload need a repository, s
 
 - **CPU**: `DarkroomEngine.process()` (`negpy/services/rendering/engine.py`) — base (geometry + normalization) → exposure (incl. dodge/burn) → clahe → lab → alt process → toning → crop → finish. The first four stages are cached per config-hash via `_run_stage()`; the rest run unconditionally. The alt-process stage (lith or cyanotype, never both) is B&W-only and off by default — when off, both engines skip it rather than run an identity pass.
 - **GPU**: `GPUEngine` (`negpy/services/rendering/gpu_engine.py`) — same logical stages as WGSL compute shaders from `negpy/features/<name>/shaders/`, with its own config-diff change detection.
-- **Orchestration**: `ImageProcessor` (`image_processor.py`) tries GPU first, falls back to CPU; export always runs full-res. `PipelineContext` carries `scale_factor`, `process_mode`, `active_roi`, and a `metrics` dict between stages.
+- **Orchestration**: `ImageProcessor` (`image_processor.py`) tries GPU first, falls back to CPU; export always runs full-res. CPU export disables stage caching with `PipelineContext.cache_stages`. Linear DNG decode, CPU saturation and unsharp masking use row blocks to bound temporary storage. `PipelineContext` carries `scale_factor`, `process_mode`, `active_roi`, and a `metrics` dict between stages.
 - **Source bakes** run before either engine, on the linear source: flat-field, sensor unmix, and every defect repair (IR, detected specks, painted heal strokes). Both engines re-upload that source per frame, so a bake reaches them parity-free and needs no shader. Each bake folds a token into `source_hash` to invalidate the engine cache.
 - **Working space**: scene-linear internally; the working OETF (Adobe RGB 1998 TRC — a pure 563/256 power, no linear segment) is applied only as the final engine step. Lab/toning compute CIELAB directly from linear, D65. Adobe RGB rather than a wide gamut because ProPhoto's imaginary primaries inflate chroma in the saturation/toning stages.
 

negpy/domain/interfaces.py

@@ -36,6 +36,7 @@ class PipelineContext:
     # As-shot WB multipliers, folded into the camera matrix when the buffer was decoded
     # without white balance (Linear RAW). None when WB was applied at decode.
     camera_wb: Optional[list] = None
+    cache_stages: bool = True
 
 
 class IImageSource(Protocol):

negpy/features/lab/logic.py

@@ -1,4 +1,5 @@
 import math
+from typing import Callable
 
 import cv2
 import numpy as np
@@ -12,6 +13,7 @@
     working_oetf_encode,
 )
 from negpy.kernel.image.validation import ensure_image
+from negpy.kernel.system.parallel import SERIAL_MAX_ELEMENTS
 
 
 CLAHE_GRID = 8
@@ -150,7 +152,31 @@ def rl_iterations(radius: float) -> int:
     return int(np.clip(int(round(10.0 * radius)), 5, 20))
 
 
-def apply_output_sharpening(
+def _map_row_blocks(img: ImageBuffer, fn: Callable[[ImageBuffer], ImageBuffer], halo: int = 0) -> ImageBuffer:
+    rows = max(1, 1048576 // img.shape[1])
+    if img.shape[0] <= rows:
+        return fn(img)
+    output = np.empty(img.shape, dtype=np.float32)
+    min_rows = math.ceil(SERIAL_MAX_ELEMENTS / img[0].size)
+    for start in range(0, img.shape[0], rows):
+        end = min(start + rows, img.shape[0])
+        low, high = max(0, start - halo), min(img.shape[0], end + halo)
+        # Keep a short tail on the full frame's Numba dispatch path.
+        low = min(low, max(0, high - min_rows))
+        block = fn(img[low:high])
+        output[start:end] = block[start - low : end - low]
+    return output
+
+
+def apply_output_sharpening(img: ImageBuffer, amount: float, radius: float = 1.0, masking: float = 0.0) -> ImageBuffer:
+    if amount <= 0:
+        return img
+    # Gaussian support, local range, and Sobel-plus-box support must cross block edges.
+    halo = max(len(gaussian_kernel_1d(radius)) // 2, 2)
+    return _map_row_blocks(img, lambda block: _output_sharpening_block(block, amount, radius, masking), halo)
+
+
+def _output_sharpening_block(
     img: ImageBuffer,
     amount: float,
     radius: float = 1.0,
@@ -236,6 +262,12 @@ def apply_rl_sharpening(
 
 
 def apply_saturation(img: ImageBuffer, saturation: float, skin_protection: float = 0.0) -> ImageBuffer:
+    if saturation == 1.0 and skin_protection <= 0.0:
+        return img
+    return _map_row_blocks(img, lambda block: _saturation_block(block, saturation, skin_protection))
+
+
+def _saturation_block(img: ImageBuffer, saturation: float, skin_protection: float = 0.0) -> ImageBuffer:
     """
     Adjusts saturation by scaling chroma (a*, b*) in CIELAB.
     Preserves perceived lightness, unlike HSV S-scaling which darkens

negpy/infrastructure/loaders/rawpy_loader.py

@@ -102,30 +102,36 @@ def tag(name: str) -> Optional[Any]:
         return None
 
     dtype_max = float(np.iinfo(arr.dtype).max) if np.issubdtype(arr.dtype, np.integer) else 1.0
-    data = arr.astype(np.float64)
-
-    if lin_table is not None:
-        idx = np.clip(data, 0, len(lin_table) - 1).astype(np.int64)
-        data = lin_table[idx]
-
     black3 = _broadcast3(black, 0.0)
     white3 = _broadcast3(white, dtype_max)
-    data = (data - black3) / np.maximum(white3 - black3, 1e-6)
-    data = np.clip(data, 0.0, 1.0)
+    denominator = np.maximum(white3 - black3, 1e-6)
 
     if len(crop_origin) >= 2 and len(crop_size) >= 2:
         ox, oy = int(round(crop_origin[0])), int(round(crop_origin[1]))
         cw, ch = int(round(crop_size[0])), int(round(crop_size[1]))
-        h, w = data.shape[:2]
+        h, w = arr.shape[:2]
         if 0 <= oy < h and 0 <= ox < w and cw > 0 and ch > 0 and (cw, ch) != (w, h):
-            data = data[oy : oy + ch, ox : ox + cw]
+            arr = arr[oy : oy + ch, ox : ox + cw]
+
+    data = np.empty(arr.shape, dtype=np.float32)
+    # Keep float64 arithmetic, but bound temporary storage to one row block.
+    rows = max(1, (8 * 1024 * 1024) // (arr.shape[1] * 3 * 8))
+    for start in range(0, arr.shape[0], rows):
+        block = arr[start : start + rows].astype(np.float64)
+        if lin_table is not None:
+            np.clip(block, 0, len(lin_table) - 1, out=block)
+            block = lin_table[block.astype(np.int64)]
+        np.subtract(block, black3, out=block)
+        np.divide(block, denominator, out=block)
+        np.clip(block, 0.0, 1.0, out=block)
+        data[start : start + rows] = block
 
     wb_gains: Optional[Tuple[float, float, float]] = None
     if len(neutral) >= 3 and all(n > 0 for n in neutral[:3]):
         r, g, b = neutral[:3]
         wb_gains = (g / r, 1.0, g / b)
 
-    return np.ascontiguousarray(data.astype(np.float32)), wb_gains
+    return data, wb_gains
 
 
 def _peek_linearraw_4ch(file_path: str) -> Optional[Tuple[np.ndarray, np.ndarray]]:

negpy/services/rendering/engine.py

@@ -52,6 +52,8 @@ def _run_stage(
         context: PipelineContext,
         pipeline_changed: bool,
     ) -> Tuple[ImageBuffer, bool]:
+        if not context.cache_stages:
+            return processor_fn(img, context), True
         conf_hash = calculate_config_hash(config)
         cached_entry = getattr(self.cache, cache_field)
 
@@ -83,6 +85,9 @@ def process(
                 process_mode=settings.process.process_mode,
             )
 
+        if not context.cache_stages:
+            self.cache.clear()
+            self._mask_plane = None
         pipeline_changed = False
         if self.cache.source_hash != source_hash:
             self.cache.clear()

negpy/services/rendering/image_processor.py

@@ -624,6 +624,7 @@ def run_pipeline(
         skip_flatfield: bool = False,
         cam_xyz: Optional[list] = None,
         camera_wb: Optional[list] = None,
+        cache_stages: bool = True,
     ) -> Tuple[Any, Dict[str, Any]]:
         """
         Executes rendering pipeline. Returns result (ndarray/GPUTexture) and metrics.
@@ -715,6 +716,7 @@ def run_pipeline(
             wants_uv_grid=wants_uv_grid,
             cam_xyz=cam_xyz,
             camera_wb=camera_wb,
+            cache_stages=cache_stages,
         )
         if metrics:
             context.metrics.update(metrics)
@@ -1228,6 +1230,7 @@ def _render_export_buffer(
                 metrics=metrics or {"log_bounds": bounds_override} if bounds_override else metrics,
                 prefer_gpu=False,
                 wants_uv_grid=False,
+                cache_stages=False,
                 skip_flatfield=True,  # f32_buffer already flat-fielded by _load_source_f32
                 cam_xyz=cam_xyz,
                 camera_wb=camera_wb,

tests/test_lab_row_blocks.py

@@ -0,0 +1,47 @@
+import numpy as np
+import pytest
+
+from negpy.features.lab.logic import (
+    apply_output_sharpening,
+    apply_saturation,
+    _output_sharpening_block,
+    _saturation_block,
+)
+from negpy.kernel.system import parallel
+
+
+@pytest.mark.parametrize("radius,mask", [(0.1, 0.0), (1.0, 0.0), (3.0, 0.6)])
+def test_sharpen_blocks_match_full_frame(radius, mask):
+    image = np.random.default_rng(3).uniform(0.01, 0.9, (1100, 1024, 3)).astype(np.float32)
+    original = image.copy()
+    full = _output_sharpening_block(image, 0.25, radius, mask)
+    blocks = apply_output_sharpening(image, 0.25, radius, mask)
+    np.testing.assert_array_equal(blocks, full)
+    np.testing.assert_array_equal(image, original)
+
+
+@pytest.mark.parametrize("parallel_enabled", [False, True])
+@pytest.mark.parametrize("tail_rows", [1, 21, 22])
+@pytest.mark.parametrize("saturation,skin", [(1.0, 1.0), (0.6, 0.7)])
+def test_saturation_short_tail_matches_full_frame(monkeypatch, parallel_enabled, tail_rows, saturation, skin):
+    monkeypatch.setattr(parallel, "_parallel_enabled", parallel_enabled)
+    image = np.random.default_rng(991).uniform(0.001, 0.999, (3072 + tail_rows, 1024, 3)).astype(np.float32)
+    image[::7] = 0.0
+    image[1::11] = 1.0
+    image = image[:, ::-1, :]
+    original = image.copy()
+    full = _saturation_block(image, saturation, skin)
+    blocks = apply_saturation(image, saturation, skin)
+    np.testing.assert_array_equal(blocks, full)
+    assert blocks.tobytes() == full.tobytes()
+    np.testing.assert_array_equal(image, original)
+
+
+@pytest.mark.parametrize("saturation,skin", [(1.0, 0.5), (0.7, 0.0), (1.3, 0.8)])
+def test_saturation_blocks_match_full_frame(saturation, skin):
+    image = np.random.default_rng(4).uniform(0.01, 0.9, (1100, 1024, 3)).astype(np.float32)
+    original = image.copy()
+    full = _saturation_block(image, saturation, skin)
+    blocks = apply_saturation(image, saturation, skin)
+    np.testing.assert_array_equal(blocks, full)
+    np.testing.assert_array_equal(image, original)

tests/test_linear_dng_memory.py

@@ -0,0 +1,49 @@
+import numpy as np
+import pytest
+import tifffile
+
+from negpy.infrastructure.loaders.rawpy_loader import _peek_linear_dng_rgb
+
+
+@pytest.mark.parametrize("use_table", [False, True])
+@pytest.mark.parametrize("crop", [None, (3, 5, 251, 1493), (0, 0, 999, 2000), (-1, 0, 10, 10)])
+def test_block_decode_matches_full_array(tmp_path, use_table, crop):
+    rng = np.random.default_rng(42)
+    codes = rng.integers(0, 4096, (1500, 257, 3), dtype=np.uint16)
+    table = np.linspace(0, 65535, 1024).astype(np.uint16)
+    tags = [
+        (50714, 5, 3, (17, 2, 256, 1, 3, 2), False),
+        (50717, 4, 3, (65535, 50000, 60000), False),
+        (50728, 5, 3, (1, 2, 1, 1, 1, 3), False),
+    ]
+    if use_table:
+        tags.append((50712, 3, len(table), tuple(table), False))
+    if crop is not None:
+        tags.extend([(50719, 10, 2, (crop[0], 1, crop[1], 1), False), (50720, 4, 2, crop[2:], False)])
+    path = tmp_path / "linear.dng"
+    tifffile.imwrite(path, codes, photometric=34892, planarconfig="contig", extratags=tags)
+    expected = codes.astype(np.float64)
+    if use_table:
+        expected = table[np.clip(expected, 0, len(table) - 1).astype(np.int64)].astype(np.float64)
+    black = np.array([8.5, 256.0, 1.5])
+    expected = np.clip((expected - black) / (np.array([65535.0, 50000.0, 60000.0]) - black), 0, 1)
+    if crop is not None:
+        x, y, w, h = crop
+        if 0 <= y < 1500 and 0 <= x < 257 and w > 0 and h > 0 and (w, h) != (257, 1500):
+            expected = expected[y : y + h, x : x + w]
+    decoded, gains = _peek_linear_dng_rgb(str(path))
+    np.testing.assert_array_equal(decoded, expected.astype(np.float32))
+    assert decoded.flags.c_contiguous
+    assert gains == (2.0, 1.0, 3.0)
+    np.testing.assert_array_equal(tifffile.imread(path), codes)
+
+
+@pytest.mark.parametrize("dtype", [np.uint8, np.uint16, np.float32, np.float64])
+def test_block_decode_default_levels(tmp_path, dtype):
+    source = np.arange(60).reshape(4, 5, 3).astype(dtype)
+    path = tmp_path / "defaults.dng"
+    tifffile.imwrite(path, source, photometric=34892, planarconfig="contig")
+    scale = np.iinfo(dtype).max if np.issubdtype(dtype, np.integer) else 1.0
+    decoded, gains = _peek_linear_dng_rgb(str(path))
+    np.testing.assert_array_equal(decoded, np.clip(source.astype(np.float64) / scale, 0, 1).astype(np.float32))
+    assert gains is None

tests/test_uncached_cpu_export.py

@@ -0,0 +1,31 @@
+from dataclasses import replace
+
+import numpy as np
+import pytest
+
+from negpy.domain.interfaces import PipelineContext
+from negpy.domain.models import WorkspaceConfig
+from negpy.features.process.models import ProcessMode
+from negpy.services.rendering.engine import DarkroomEngine
+
+
+@pytest.mark.parametrize("mode", list(ProcessMode))
+def test_uncached_render_matches_cached_and_leaves_no_stage_arrays(mode):
+    source = np.random.default_rng(91).uniform(0.02, 0.85, (48, 64, 3)).astype(np.float32)
+    original = source.copy()
+    config = WorkspaceConfig()
+    config = replace(config, process=replace(config.process, process_mode=mode))
+    engine = DarkroomEngine()
+    cached = engine.process(source, config, "same-source")
+    assert engine.cache.base is not None
+    context = PipelineContext(
+        original_size=(48, 64), scale_factor=64 / engine.config.preview_render_size, process_mode=mode, cache_stages=False
+    )
+    uncached = engine.process(source, config, "same-source", context)
+    np.testing.assert_array_equal(cached, uncached)
+    np.testing.assert_array_equal(source, original)
+    for name in ("base", "exposure", "clahe", "lab"):
+        assert getattr(engine.cache, name) is None
+    preview = engine.process(source, config, "same-source")
+    np.testing.assert_array_equal(preview, cached)
+    assert engine.cache.base is not None