Optimize DICOM UPP registration lazy loading and fusion views

DICOM_and_UPP配准/app.py

@@ -12,6 +12,7 @@ import shutil
 import subprocess
 import threading
 import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from collections import Counter, defaultdict
 from pathlib import Path
 from typing import Any
@@ -123,6 +124,7 @@ app = FastAPI(title="DICOM UPP Registration Workspace")
 app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
 TOKENS: dict[str, str] = {}
 STUDY_CACHE: dict[str, dict[str, Any]] = {}
+SERIES_SORT_CACHE: dict[str, tuple[float, list[Path]]] = {}
 STACK_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
 CACHE_REFRESH_LOCK = threading.Lock()
 CACHE_REFRESH_THREAD: threading.Thread | None = None
@@ -663,9 +665,10 @@ def cases(
     status_filter: str = Query(default="", alias="status"),
     body_part: str = "",
     algorithm_model: str = "",
-    limit: int = Query(default=120, ge=1, le=400),
+    limit: int = Query(default=20, ge=1, le=80),
+    offset: int = Query(default=0, ge=0),
     _: str = Depends(require_auth),
-) -> list[dict[str, Any]]:
+) -> dict[str, Any]:
     ensure_registration_table()
     clauses = []
     if q.strip():
@@ -690,7 +693,15 @@ def cases(
     if algorithm_model.strip():
         clauses.append(f"algorithm_model ILIKE {sql_literal('%' + algorithm_model.strip().replace('%', '').replace('_', '') + '%')}")
     where_sql = "WHERE " + " AND ".join(clauses) if clauses else ""
-    return pg_json_rows(
+    total = int(pg_scalar(
+        f"""
+        SELECT count(*)::int
+        FROM ({relation_select()}) relation
+        {where_sql}
+        """,
+        timeout=20,
+    ) or 0)
+    rows = pg_json_rows(
         f"""
         SELECT *
         FROM ({relation_select()}) relation
@@ -701,9 +712,17 @@ def cases(
             COALESCE(study_time, '') DESC,
             ct_key
         LIMIT {int(limit)}
+        OFFSET {int(offset)}
         """,
         timeout=20,
     )
+    return {
+        "rows": rows,
+        "total": total,
+        "limit": int(limit),
+        "offset": int(offset),
+        "has_more": int(offset) + len(rows) < total,
+    }
 
 
 @app.get("/api/cases/{ct_number}")
@@ -739,6 +758,23 @@ def read_header(path: Path) -> dict[str, str]:
     return {tag: text(getattr(ds, tag, "")) for tag in DICOM_TAGS}
 
 
+def read_headers_bulk(paths: list[Path], max_workers: int = 10) -> dict[Path, dict[str, str]]:
+    unique_paths = list(dict.fromkeys(paths))
+    if not unique_paths:
+        return {}
+    results: dict[Path, dict[str, str]] = {}
+    workers = max(1, min(max_workers, len(unique_paths)))
+    with ThreadPoolExecutor(max_workers=workers) as executor:
+        future_map = {executor.submit(read_header, path): path for path in unique_paths}
+        for future in as_completed(future_map):
+            path = future_map[future]
+            try:
+                results[path] = future.result()
+            except Exception:
+                continue
+    return results
+
+
 def sort_key(item: tuple[Path, dict[str, str]]) -> tuple[float, float, str]:
     path, meta = item
     z = numeric(meta.get("SliceLocation"), 0.0)
@@ -823,29 +859,36 @@ def scan_study(ct_number: str) -> dict[str, Any]:
     root = resolve_study_root(study)
     if not root.exists():
         raise HTTPException(status_code=404, detail=f"DICOM 目录不存在：{root}")
-    grouped: dict[str, list[tuple[Path, dict[str, str]]]] = defaultdict(list)
+    grouped: dict[str, list[Path]] = defaultdict(list)
     for path in root.rglob("*.dcm"):
-        try:
-            meta = read_header(path)
-        except Exception:
-            continue
-        uid = meta.get("SeriesInstanceUID") or path.parent.name
-        grouped[uid].append((path, meta))
+        grouped[path.parent.name].append(path)
     annotations = get_annotations(ct_number)
     series = []
     file_map = {}
-    for uid, items in grouped.items():
-        items.sort(key=sort_key)
-        first = items[0][1]
-        last = items[-1][1]
-        file_map[uid] = [path for path, _ in items]
-        annotation = annotations.get(uid) or {}
+    summary_paths: list[Path] = []
+    for paths in grouped.values():
+        paths = sorted(paths)
+        if not paths:
+            continue
+        summary_paths.append(paths[0])
+    header_map = read_headers_bulk(summary_paths)
+    for uid, paths in grouped.items():
+        paths = sorted(paths)
+        if not paths:
+            continue
+        first = header_map.get(paths[0])
+        if not first:
+            continue
+        last = first
+        actual_uid = first.get("SeriesInstanceUID") or uid
+        file_map[actual_uid] = paths
+        annotation = annotations.get(actual_uid) or annotations.get(uid) or {}
         series.append(
             {
-                "series_uid": uid,
+                "series_uid": actual_uid,
                 "description": first.get("SeriesDescription") or "未命名序列",
                 "series_number": first.get("SeriesNumber") or "",
-                "count": len(items),
+                "count": len(paths),
                 "modality": first.get("Modality") or "",
                 "rows": first.get("Rows") or "",
                 "columns": first.get("Columns") or "",
@@ -876,7 +919,22 @@ def get_series_files(ct_number: str, series_uid: str) -> list[Path]:
     files = data["files"].get(series_uid)
     if not files:
         raise HTTPException(status_code=404, detail="DICOM 序列不存在")
-    return files
+    cache_key = f"{normalize_ct(ct_number)}|{series_uid}"
+    cached = SERIES_SORT_CACHE.get(cache_key)
+    if cached:
+        SERIES_SORT_CACHE[cache_key] = (time.time(), cached[1])
+        return cached[1]
+    header_map = read_headers_bulk(list(files), max_workers=12)
+    items: list[tuple[Path, dict[str, str]]] = [(path, meta) for path, meta in header_map.items()]
+    if not items:
+        raise HTTPException(status_code=404, detail="DICOM 序列文件不可读")
+    sorted_files = [path for path, _ in sorted(items, key=sort_key)]
+    data["files"][series_uid] = sorted_files
+    SERIES_SORT_CACHE[cache_key] = (time.time(), sorted_files)
+    if len(SERIES_SORT_CACHE) > 8:
+        oldest = sorted(SERIES_SORT_CACHE.items(), key=lambda item: item[1][0])[0][0]
+        SERIES_SORT_CACHE.pop(oldest, None)
+    return sorted_files
 
 
 def window_values(ds: pydicom.Dataset, preset: str) -> tuple[float, float]:
@@ -945,7 +1003,7 @@ def render_array(arr: np.ndarray, center: float, width: float, max_size: int = 9
     pil = Image.fromarray(img)
     pil = resize_for_spacing(pil, spacing[0], spacing[1])
     if max(pil.size) > max_size:
-        pil.thumbnail((max_size, max_size), Image.Resampling.BILINEAR)
+        pil.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)
     output = io.BytesIO()
     pil.save(output, format="PNG", optimize=True)
     return output.getvalue()
@@ -1016,7 +1074,7 @@ def dicom_fusion_volume(
     else:
         start = max(0, center_index - radius)
         end = min(total - 1, center_index + radius)
-    max_frames = {"low": 48, "medium": 64, "high": 80}.get(str(detail).lower(), 64)
+    max_frames = {"low": 24, "medium": 48, "high": 72}.get(str(detail).lower(), 48)
     if end - start + 1 > max_frames:
         step = int(np.ceil((end - start + 1) / max_frames))
         indices = list(range(start, end + 1, step))
@@ -1226,6 +1284,7 @@ def run_cache_refresh(reason: str = "manual", user: str = "system") -> None:
     try:
         cache_root.mkdir(parents=True, exist_ok=True)
         STUDY_CACHE.clear()
+        SERIES_SORT_CACHE.clear()
         STACK_CACHE.clear()
         rows = relation_rows_for_cache()
         total = len(rows)