refactor!: 降噪回归解码链语义——原生分辨率一次,视图派生,缩放永不重算

用户架构裁定:降噪就是去马赛克后的一环,做一次就不该再被触碰;
分级只属于下游视图。撤销 proxy/view 两个临时分级:
- _executor_denoise 重写:无论首个视图是什么,都对原生全分辨率源
  计算一次(内存+磁盘双缓存,键含模型版本);代理视图拿降采样、
  缩放 ROI 直接裁 full、导出用 full——零重算。
- 计算期间仅当切换到"另一张照片"时让路(progress 回调比较 pending
  path);同照片的调参请求排队等待,不再因中断反复重来。
- 新增跨会话磁盘缓存 denoise_disk_cache(fp16+zstd ≈150MB/42MP,
  LRU 上限 RAWALCHEMY_DENOISE_CACHE_GB 默认 20)——重开照片秒载入。
- SCUNet 42s 是每张照片一生只付一次的成本;FastDenoise 落地后同一
  架构直接把这一次缩到 ~2.4s。
- 241 测试全过(新契约:原生一次+全视图派生、跨会话磁盘命中)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: shenmintao

src/raw_alchemy/pipeline/denoise_disk_cache.py

@@ -0,0 +1,109 @@
+"""降噪结果磁盘缓存:全图降噪一次付费、跨会话免费。
+
+键:RAW 路径 + 文件大小 + mtime + 降噪模型文件名(模型升级自动失效)。
+存储:线性工作空间 float16 + zstd(≈150MB/42MP 张),LRU 按 atime 逐出。
+配置:RAWALCHEMY_DENOISE_CACHE_DIR / RAWALCHEMY_DENOISE_CACHE_GB(默认 20)。
+"""
+
+import hashlib
+import os
+import time
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+import zstandard
+from loguru import logger
+
+_MAGIC = b"RADC1\n"
+
+
+def _cache_dir() -> Path:
+    d = os.environ.get("RAWALCHEMY_DENOISE_CACHE_DIR")
+    p = Path(d) if d else Path.home() / ".rawalchemy" / "denoise_cache"
+    p.mkdir(parents=True, exist_ok=True)
+    return p
+
+
+def _limit_bytes() -> int:
+    try:
+        gb = float(os.environ.get("RAWALCHEMY_DENOISE_CACHE_GB", "20"))
+    except ValueError:
+        gb = 20.0
+    return int(gb * (1 << 30))
+
+
+def _key(raw_path: str, model_tag: str) -> Optional[str]:
+    try:
+        st = os.stat(raw_path)
+    except OSError:
+        return None
+    h = hashlib.sha1(
+        f"{os.path.abspath(raw_path)}|{st.st_size}|{int(st.st_mtime)}|{model_tag}"
+        .encode("utf-8", "replace")
+    ).hexdigest()
+    return h
+
+
+def load(raw_path: str, model_tag: str) -> Optional[np.ndarray]:
+    """命中返回 (H, W, 3) float32 线性工作空间;未命中返回 None。"""
+    k = _key(raw_path, model_tag)
+    if k is None:
+        return None
+    f = _cache_dir() / f"{k}.radc"
+    if not f.exists():
+        return None
+    try:
+        t0 = time.time()
+        blob = f.read_bytes()
+        assert blob[:6] == _MAGIC
+        h, w = int.from_bytes(blob[6:10], "little"), int.from_bytes(blob[10:14], "little")
+        data = zstandard.ZstdDecompressor().decompress(blob[14:], max_output_size=h * w * 3 * 2)
+        arr = np.frombuffer(data, np.float16).reshape(h, w, 3).astype(np.float32)
+        os.utime(f)  # LRU: 命中刷新 atime/mtime
+        logger.info(f"[DenoiseCache] hit {os.path.basename(raw_path)} "
+                    f"({time.time() - t0:.2f}s load)")
+        return arr
+    except Exception as e:
+        logger.warning(f"[DenoiseCache] corrupt entry dropped: {e}")
+        try:
+            f.unlink()
+        except OSError:
+            pass
+        return None
+
+
+def save(raw_path: str, model_tag: str, denoised: np.ndarray) -> None:
+    k = _key(raw_path, model_tag)
+    if k is None:
+        return
+    try:
+        t0 = time.time()
+        h, w = denoised.shape[:2]
+        payload = zstandard.ZstdCompressor(level=1).compress(
+            np.ascontiguousarray(denoised, np.float32).astype(np.float16).tobytes())
+        blob = _MAGIC + h.to_bytes(4, "little") + w.to_bytes(4, "little") + payload
+        f = _cache_dir() / f"{k}.radc"
+        tmp = f.with_suffix(".tmp")
+        tmp.write_bytes(blob)
+        tmp.replace(f)
+        logger.info(f"[DenoiseCache] saved {os.path.basename(raw_path)} "
+                    f"({len(blob) / 1e6:.0f}MB, {time.time() - t0:.2f}s)")
+        _evict()
+    except Exception as e:
+        logger.warning(f"[DenoiseCache] save failed: {e}")
+
+
+def _evict() -> None:
+    limit = _limit_bytes()
+    d = _cache_dir()
+    files = sorted(d.glob("*.radc"), key=lambda f: f.stat().st_mtime)
+    total = sum(f.stat().st_size for f in files)
+    while total > limit and files:
+        f = files.pop(0)
+        try:
+            total -= f.stat().st_size
+            f.unlink()
+            logger.info(f"[DenoiseCache] evicted {f.name}")
+        except OSError:
+            break

src/raw_alchemy/pipeline/ops.py

@@ -47,10 +47,7 @@ def build_op_list(params: ProcessorParams) -> list[Op]:
     working_space = config.WORKING_SPACE
 
     if params.get("denoise_enabled", False):
-        # tier: 'proxy' while only the fast proxy-resolution denoise exists,
-        # 'full' once the full-res result is cached (worker injects it) —
-        # keying prefix/output caches so tier upgrades re-render cleanly.
-        ops.append(Op("denoise", (params.get("_denoise_tier"),)))
+        ops.append(Op("denoise", ()))
 
     if params.get("lens_correct", False):
         ops.append(Op("lens_correct", (_as_hashable(params.get("custom_db_path")),)))

src/raw_alchemy/workers/image_processor.py

@@ -129,12 +129,6 @@ def __init__(self):
         self.cached_denoise_full = None
         self.cached_denoise_proxy = None
         self.last_denoise_key = None
-        # Fast tier: proxy-resolution denoise (~3s) shown while the full-res
-        # pass (~40s, abortable, idle-scheduled) hasn't landed yet.
-        self.last_denoise_proxy_key = None
-        # View tier: per-ROI denoise results (zoomed views before the full
-        # pass lands). Keyed by (path, roi_rect, shape); tiny LRU.
-        self._denoise_view_cache: dict = {}
 
         self._should_stop = False
         self._gpu_uint8 = None      # Pre-allocated pooled uint8 output buffer
@@ -1045,130 +1039,70 @@ def _trim_executor_caches(self, budget_bytes: Optional[int] = None):
             full.trim(remaining)
 
     def _executor_denoise(self, src: np.ndarray) -> np.ndarray:
+        """降噪 = 解码链的一环(去马赛克后):原生分辨率只算一次,
+        内存+磁盘双缓存;所有视图(代理/任意缩放 ROI/导出)从结果派生,
+        缩放拖动永不触发重算。计算期间仅当切换到其他照片时让路。"""
         path = self._executor_path
         if path is None:
             return src
 
         denoise_cache_key = (path, 'denoise')
-
-        # Proxy-mode preview. With the full result cached, serve its
-        # downscale; otherwise denoise the proxy itself (~3s) as the fast
-        # tier — the full pass runs later (idle refine / export) and its
-        # arrival flips the op tier, re-rendering from the exact result.
-        if self._executor_using_proxy:
-            if (
-                denoise_cache_key == self.last_denoise_key
-                and self.cached_denoise_full is not None
-            ):
-                if self.cached_denoise_proxy is None:
-                    self.cached_denoise_proxy = self._make_proxy(self.cached_denoise_full)
-                if self.cached_denoise_proxy is not None:
-                    return np.ascontiguousarray(self.cached_denoise_proxy)
-            proxy_key = (path, 'denoise-proxy')
-            if (
-                proxy_key == self.last_denoise_proxy_key
-                and self.cached_denoise_proxy is not None
-            ):
-                return np.ascontiguousarray(self.cached_denoise_proxy)
-            try:
-                self.denoise_started.emit()
-                logger.info("[Worker] SCUNet proxy-tier denoise (fast preview)...")
-
-                def progress_cb(cur, total):
-                    self.denoise_progress.emit(cur, total)
-
-                denoised = denoise_rgb_linear(src, progress_callback=progress_cb)
-                self.cached_denoise_proxy = denoised
-                self.last_denoise_proxy_key = proxy_key
-                self.denoise_finished.emit()
-                return np.ascontiguousarray(denoised)
-            except Exception as e:
-                logger.error(f"[Worker] Proxy denoise failed: {e}")
-                self.denoise_finished.emit()
-                return src
-            finally:
-                denoise_clear_session()
-
-        if self._executor_params.get('_denoise_tier') == 'view':
-            rect = self._executor_params.get('_preview_roi')
-            vkey = (path, tuple(rect) if rect else None, src.shape)
-            cached = self._denoise_view_cache.get(vkey)
+        missing = (denoise_cache_key != self.last_denoise_key
+                   or self.cached_denoise_full is None)
+        if missing:
+            from raw_alchemy.pipeline import denoise_disk_cache
+            from raw_alchemy.onnx.rgb_denoiser import MODEL_FILE as _DN_TAG
+            cached = denoise_disk_cache.load(path, _DN_TAG)
             if cached is not None:
-                return np.ascontiguousarray(cached)
-            try:
-                self.denoise_started.emit()
-                logger.info("[Worker] SCUNet view-tier denoise (ROI)...")
-                abortable = bool(self._executor_params.get('_denoise_abortable'))
-
-                def progress_cb(cur, total):
-                    self.denoise_progress.emit(cur, total)
-                    if abortable:
-                        with self.lock:
-                            superseded = self.pending_request is not None
-                        if superseded:
-                            raise PipelineAborted(
-                                f"view denoise superseded at tile {cur}/{total}"
-                            )
-
-                denoised = denoise_rgb_linear(src, progress_callback=progress_cb)
-                if len(self._denoise_view_cache) >= 4:
-                    self._denoise_view_cache.pop(next(iter(self._denoise_view_cache)))
-                self._denoise_view_cache[vkey] = denoised
-                self.denoise_finished.emit()
-                return np.ascontiguousarray(denoised)
-            except PipelineAborted:
-                self.denoise_finished.emit()
-                raise
-            except Exception as e:
-                logger.error(f"[Worker] View denoise failed: {e}")
-                self.denoise_finished.emit()
-                return src
-            finally:
-                denoise_clear_session()
+                self.cached_denoise_original = self.cpu_linear
+                self.cached_denoise_full = cached
+                self.cached_denoise_proxy = None
+                self.last_denoise_key = denoise_cache_key
+                missing = False
 
-        if denoise_cache_key != self.last_denoise_key or self.cached_denoise_full is None:
+        if missing:
+            # 无论当前视图是代理还是 ROI,都对原生全分辨率源计算
+            source = self.cpu_linear if self.cpu_linear is not None else src
             try:
                 self.denoise_started.emit()
-                logger.info("[Worker] SCUNet RGB denoise (post-demosaic)...")
-
-                abortable = bool(self._executor_params.get('_denoise_abortable'))
+                logger.info("[Worker] SCUNet RGB denoise (native, once)...")
 
                 def progress_cb(cur, total):
                     self.denoise_progress.emit(cur, total)
-                    if abortable:
-                        with self.lock:
-                            superseded = self.pending_request is not None
-                        if superseded:
-                            raise PipelineAborted(
-                                f"full denoise superseded at tile {cur}/{total}"
-                            )
-
-                denoised = denoise_rgb_linear(
-                    src,
-                    progress_callback=progress_cb,
-                )
+                    with self.lock:
+                        pr = self.pending_request
+                    if pr is not None and pr.path != path:
+                        raise PipelineAborted(
+                            f"denoise superseded by another photo at {cur}/{total}")
 
+                denoised = denoise_rgb_linear(source, progress_callback=progress_cb)
                 self.cached_denoise_original = self.cpu_linear
                 self.cached_denoise_full = denoised
                 self.cached_denoise_proxy = None
                 self.last_denoise_key = denoise_cache_key
+                denoise_disk_cache.save(path, _DN_TAG, denoised)
                 self.denoise_finished.emit()
             except PipelineAborted:
                 self.denoise_finished.emit()
                 raise
             except Exception as e:
-                logger.error(f"[Worker] Denoising failed: {e}")
+                logger.error(f"[Worker] Denoising failed: {type(e).__name__}: {e}")
                 self.denoise_finished.emit()
                 self.cached_denoise_original = None
                 self.cached_denoise_full = None
                 self.cached_denoise_proxy = None
                 self.last_denoise_key = None
                 return src
             finally:
-                # Always drop the ONNX session: on DirectML it pins hundreds
-                # of MB of VRAM that the interactive preview renderer needs.
                 denoise_clear_session()
 
+        if self._executor_using_proxy:
+            if self.cached_denoise_proxy is None:
+                self.cached_denoise_proxy = self._make_proxy(self.cached_denoise_full)
+            if self.cached_denoise_proxy is not None:
+                return np.ascontiguousarray(self.cached_denoise_proxy)
+            return np.ascontiguousarray(self.cached_denoise_full)
+
         denoised = np.ascontiguousarray(self.cached_denoise_full)
         if not self._executor_params.get('lens_correct', False):
             self.cpu_corrected = denoised
@@ -1296,16 +1230,11 @@ def _executor_auto_gain(self, _metering_img: np.ndarray, metering_mode: str) ->
     def _prepare_executor_source_state(self, params: ProcessorParams):
         self._executor_corrected_source = None
         denoise_enabled = params.get('denoise_enabled', False)
-        if not denoise_enabled and (
-            self.last_denoise_key is not None
-            or self.last_denoise_proxy_key is not None
-        ):
+        if not denoise_enabled and self.last_denoise_key is not None:
             self.cached_denoise_original = None
             self.cached_denoise_full = None
             self.cached_denoise_proxy = None
             self.last_denoise_key = None
-            self.last_denoise_proxy_key = None
-            self._denoise_view_cache.clear()
 
         if not params.get('lens_correct', False):
             if denoise_enabled and self.cached_denoise_full is not None:
@@ -1467,25 +1396,6 @@ def _do_process(self, request: ProcessRequest):
             # zoom>fit ROI rendering (T7.5): resolve the visible-region crop
             # before the output key so the view key covers the ROI rect.
             roi_info = None if use_proxy else self._compute_preview_roi(params)
-            if params.get('denoise_enabled', False):
-                full_ready = (
-                    (request.path, 'denoise') == self.last_denoise_key
-                    and self.cached_denoise_full is not None
-                )
-                # tier(与 DPI 挡位同一哲学:屏幕需要多少就处理多少):
-                #   proxy = fit 视图对 3MP 代理降噪(~3s)
-                #   view  = 放大视图只对 DPI 挡位后的 ROI 降噪(~6-19s,带缓存)
-                #   full  = 精确结果(空闲精化产出/导出)
-                if full_ready:
-                    params['_denoise_tier'] = 'full'
-                elif use_proxy:
-                    params['_denoise_tier'] = 'proxy'
-                elif roi_info is not None:
-                    params['_denoise_tier'] = 'view'
-                else:
-                    params['_denoise_tier'] = 'full'
-                # 预览请求中的降噪按 tile 让路给新交互(空闲精化重试)
-                params['_denoise_abortable'] = True
             if roi_info is not None:
                 params['_preview_roi'] = roi_info[0]
                 params['_roi_full_size'] = roi_info[1]
@@ -1536,13 +1446,6 @@ def _do_process(self, request: ProcessRequest):
                 _tw, _th = self._make_roi_target_size(_rw, _rh, _fw, _fh, params)
                 _roi_out = (_tw, _th) if (_tw < _rw * 0.95 or _th < _rh * 0.95) else None
                 ops = self._insert_roi_op(ops, roi_info[0], _roi_out)
-                if params.get('_denoise_tier') == 'view':
-                    di = next((k for k, o in enumerate(ops) if o.name == 'denoise'), None)
-                    ri = next((k for k, o in enumerate(ops) if o.name == 'roi'), None)
-                    if di is not None and ri is not None and di < ri:
-                        d_op = ops.pop(di)
-                        ri = next(k for k, o in enumerate(ops) if o.name == 'roi')
-                        ops.insert(ri + 1, d_op)
             executor_source = self._select_executor_source(use_proxy)
             executor = self._get_preview_executor()
             # Cooperative cancellation (T7.3): a newer user request aborts