fix(transparency): meter the picture box of an uncropped slide on the CPU

negpy/features/transparency/processor.py

@@ -92,7 +92,7 @@ def process(self, image: ImageBuffer, context: PipelineContext) -> ImageBuffer:
         # working-space (post camera matrix) and post-unmix, since that is what this curve
         # itself consumes.
         an_roi, an_buffer = resolve_analysis_region(
-            linear.shape, context.active_roi, self.config.analysis_buffer, self.config.analysis_rect
+            linear.shape, context.active_roi, self.config.analysis_buffer, self.config.analysis_rect, self.config.meter_rect
         )
         prefiltered = unmix_log_image(prefilter_log_grid(linear, an_roi, an_buffer), unmix)
 
@@ -117,7 +117,7 @@ def process(self, image: ImageBuffer, context: PipelineContext) -> ImageBuffer:
         context.metrics["final_bounds"] = bounds
         context.metrics["norm_density_range"] = luminance_density_range(bounds)
         context.metrics["density_above_base"] = None  # a slide has no clear base to read above
-        context.metrics["scan_clip_fractions"] = measure_clip_fractions(image, context.active_roi, self.config.analysis_buffer)
+        context.metrics["scan_clip_fractions"] = measure_clip_fractions(image, an_roi, an_buffer)
         context.metrics["normalized_log"] = res
         context.metrics["histogram_density"] = density_histogram(res, context.active_roi)
         # The Crop tool shows the whole frame, so the parade's columns span it too (the GPU's full_frame).

negpy/infrastructure/gpu/shader_loader.py

@@ -11,11 +11,17 @@ class ShaderLoader:
     """
 
     _cache: dict[str, Any] = {}
+    # A module belongs to the device that compiled it: a new device after destroy_singleton empties the cache.
+    _device: Any = None
     _lock = threading.Lock()
 
     @classmethod
     def load(cls, path: str) -> Any:
         with cls._lock:
+            gpu = GPUDevice.get()
+            if gpu.device is not cls._device:
+                cls._cache.clear()
+                cls._device = gpu.device
             if path in cls._cache:
                 return cls._cache[path]
 
@@ -25,7 +31,6 @@ def load(cls, path: str) -> Any:
             with open(path, "r", encoding="utf-8") as f:
                 code = f.read()
 
-            gpu = GPUDevice.get()
             if not gpu.device:
                 raise RuntimeError("Hardware device required for shader compilation")
 

negpy/services/assets/thumbnail_fingerprint.py

@@ -12,7 +12,7 @@
 from negpy.domain.models import WorkspaceConfig
 
 # Bump in any change that alters rendered pixels for unchanged settings, a changed default included.
-THUMBNAIL_RENDER_VERSION = 29
+THUMBNAIL_RENDER_VERSION = 30
 
 # Marks a thumbnail made from the source preview, which runs none of the frame's settings.
 QUICK = "quick"

tests/test_gpu_infrastructure.py

@@ -29,6 +29,20 @@ def test_shader_loader(self):
             module2 = ShaderLoader.load(shader_path)
             self.assertIs(module, module2)
 
+    def test_shader_loader_follows_a_new_device(self):
+        """A module compiled on a destroyed device is never handed to the next one."""
+        import os
+
+        if not self.gpu.is_available:
+            self.skipTest("GPU not available")
+        shader_path = os.path.join("negpy", "features", "geometry", "shaders", "transform.wgsl")
+        before = ShaderLoader.load(shader_path)
+        GPUDevice.destroy_singleton()
+        device = GPUDevice.get().device
+        after = ShaderLoader.load(shader_path)
+        self.assertIsNot(before, after)
+        device.create_compute_pipeline(layout="auto", compute={"module": after, "entry_point": "main"})
+
     def test_gpu_texture(self):
         """GPUTexture should initialize and upload data."""
         if not self.gpu.is_available:

tests/test_transparency_meter_rect.py

@@ -0,0 +1,85 @@
+"""The transparency path meters the picture box (`meter_rect`) of a frame with no crop, on both engines."""
+
+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, with_positive_source, with_process_mode
+from negpy.services.rendering.engine import base_processor
+
+_H, _W = 240, 360
+#: (x1, y1, x2, y2) of the picture box, normalized.
+_BOX = (0.25, 0.25, 0.75, 0.75)
+
+
+def _wedge(h: int, w: int, cast: tuple) -> np.ndarray:
+    v = np.geomspace(5e-4, 0.9, h * w).astype(np.float32).reshape(h, w)
+    return np.stack([v * cast[0], v * cast[1], v * cast[2]], axis=-1)
+
+
+def _framed_slide() -> np.ndarray:
+    """A warm-cast wedge inside a surround with the opposite cast: the axis says which one was metered."""
+    rng = np.random.default_rng(3)
+    img = _wedge(_H, _W, (0.55, 0.8, 1.0))
+    y1, y2, x1, x2 = int(_BOX[1] * _H), int(_BOX[3] * _H), int(_BOX[0] * _W), int(_BOX[2] * _W)
+    img[y1:y2, x1:x2] = _wedge(y2 - y1, x2 - x1, (1.0, 0.82, 0.62))
+    # Clipped red outside the box and clipped blue inside it: the clip fractions say which region was read.
+    img[: _H // 8, : _W // 8, 0] = 1.0
+    img[y1 : y1 + _H // 16, x1 : x1 + _W // 16, 2] = 1.0
+    return np.ascontiguousarray(img + rng.uniform(0, 1e-4, img.shape).astype(np.float32))
+
+
+def _config(meter_rect) -> WorkspaceConfig:
+    cfg = with_positive_source(with_process_mode(WorkspaceConfig(), ProcessMode.E6), True)
+    return replace(
+        cfg,
+        geometry=replace(cfg.geometry, autocrop_offset=0),
+        exposure=replace(cfg.exposure, cast_removal_strength=1.0),
+        process=replace(cfg.process, meter_rect=meter_rect),
+    )
+
+
+def _cpu_metrics(img: np.ndarray, cfg: WorkspaceConfig) -> dict:
+    ctx = PipelineContext(scale_factor=1.0, original_size=img.shape[:2], process_mode=cfg.process.process_mode)
+    base_processor(cfg).process(img, ctx)
+    return ctx.metrics
+
+
+def test_the_cpu_axis_reads_only_the_picture_box():
+    img = _framed_slide()
+    boxed = _cpu_metrics(img, _config(_BOX))["neutral_axis_refs"]
+    whole = _cpu_metrics(img, _config(None))["neutral_axis_refs"]
+    assert boxed is not None and whole is not None
+    assert not np.allclose(boxed[0], whole[0], atol=1e-3)
+
+
+def test_the_cpu_clip_fractions_read_only_the_picture_box():
+    img = _framed_slide()
+    boxed = _cpu_metrics(img, _config(_BOX))["scan_clip_fractions"]
+    whole = _cpu_metrics(img, _config(None))["scan_clip_fractions"]
+    assert boxed[0] == 0.0 and boxed[2] > 0.0
+    assert whole[0] > 0.0
+
+
+def test_the_gpu_meters_the_same_axis_as_the_cpu():
+    from negpy.infrastructure.gpu.device import GPUDevice
+
+    if not GPUDevice.get().is_available:
+        pytest.skip("GPU not available")
+    from negpy.kernel.system.config import render_scale_factor
+    from negpy.services.rendering.gpu_engine import GPUEngine
+
+    img = _framed_slide()
+    cfg = _config(_BOX)
+    engine = GPUEngine()
+    try:
+        _, gpu = engine.process_to_texture(img, cfg, scale_factor=render_scale_factor(img.shape), apply_layout=False, readback_metrics=True)
+    finally:
+        engine.destroy_all()
+    cpu = _cpu_metrics(img, cfg)
+    for ref in (0, 1):
+        np.testing.assert_allclose(gpu["neutral_axis_refs"][ref], cpu["neutral_axis_refs"][ref], atol=1e-5)
+    np.testing.assert_allclose(gpu["scan_clip_fractions"], cpu["scan_clip_fractions"], atol=1e-6)

tests/thumbnail_fingerprint_defaults.json

@@ -211,5 +211,5 @@
    "vanadium_strength": "0.0"
   }
  },
- "render_version": 29
+ "render_version": 30
 }