fix(ceus): 降低长视频内存峰值,快照开跑即写,build 核对 CEUS 运行与交付包

- 内存:prepare_segmentation 改为流式解码,只留分屏和配准的抽样帧,绿框逐帧在两半上找;
  分割那一半由 SegmentationInput.target_frames() 到用时再解码,segment_video 边读边写 jpg;
  MedSAM2 init_state 的读帧换成直接在存放设备上建张量(不改 MedSAM2 源码,结果逐位相同);
  传播结果写进预分配数组。帧数最多的一段视频常驻内存峰值 11.2 GiB → 2.3 GiB,分辨率最大的
  一段原代码被 OOM 杀掉、现为 4.2 GiB;两段的掩码和全部产物与改前逐字节相同,33 段视频的
  第 1–2 步结果与改前一致,29 例配准与 review.csv 一致。
- 快照:ceus_quant 开跑时写 snapshot/(git、ceus.yaml、review.csv、uv.lock、segment.json),
  配置和核对表从快照副本读;收尾由新函数 runs.finish_snapshot 只新建 finished.json。
  build 与 ceus_align_review 同样改用 finish_snapshot 记结束时间。
- build:summary.json 加 video_sha256;装 CEUS 结果时核对快照里的 delivery_dir、逐例核对
  视频 SHA-256 和推注时间,不同就整次停下;新字段做类型检查,不进 tic.json。

来源:主会话任务(问题 1–3 及补充的 delivery_dir 比对)。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-16 09:01:44 -04:00
parent e165b32b9d
commit 1de88a5d0a
23 changed files with 561 additions and 202 deletions
+7 -3
View File
@@ -27,7 +27,7 @@ from cadx.delivery import (
review_sentences,
)
from cadx.library import LibraryCase, LibraryCeus, check_key_image, write_library
from cadx.runs import make_build_id, read_git_state, write_snapshot
from cadx.runs import finish_snapshot, make_build_id, read_git_state, write_snapshot
ALIAS_ALPHABET = string.ascii_lowercase + string.digits
ALIAS_LENGTH = 4
@@ -155,7 +155,11 @@ def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary
delivery = read_delivery(config.delivery_dir, code_maps)
passed, key_image_skips = _check_key_images(delivery.cases, config.delivery_dir)
skipped = [*delivery.skipped, *key_image_skips]
ceus_run = None if config.ceus_run is None else open_ceus_run(config.ceus_run)
ceus_run = (
None
if config.ceus_run is None
else open_ceus_run(config.ceus_run, config.delivery_dir, repo_root)
)
ceus_of: dict[str, LibraryCeus | None] = dict.fromkeys((c.case_no for c in passed), None)
if ceus_run is not None:
with_ceus, ceus_skips = _attach_ceus(passed, ceus_run)
@@ -214,10 +218,10 @@ def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary
build_id=build_id,
git=git,
started_at=now,
finished_at=datetime.now().astimezone(),
uv_lock=repo_root / "uv.lock",
configs=configs,
)
finish_snapshot(answers_partial / "snapshot", datetime.now().astimezone(), {})
write_answers(
answers_dir=answers_partial,
build_id=build_id,
+31 -4
View File
@@ -1,16 +1,20 @@
"""读一次 CEUS 定量运行(`runs/ceus/<运行号>/`),按白名单挑出超声一路的 tic.json 和关键帧(C15)。
运行目录的布局由 `scripts/ceus_quant.py` 定:`failed.csv`、`snapshot/snapshot.json`、
`<视频号>/summary.json`、`<视频号>/keyframes/<槽位>.png`。tic.json 的字段只在这里列,
视频号、路径、配置、帧号和拟合参数都不进去。
`snapshot/ceus.yaml`、`<视频号>/summary.json`、`<视频号>/keyframes/<槽位>.png`。tic.json 的字段
只在这里列,视频号、路径、配置、帧号和拟合参数都不进去。快照里的交付包路径、summary.json 里记的
视频 SHA-256 和推注时间要与这份交付包一致,否则运行不是在这份交付包上跑的,整次停下。
"""
import csv
import hashlib
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import yaml
from cadx.delivery import CaseRecord, SkipRecord
from cadx.library import LibraryCeus
@@ -51,14 +55,27 @@ class CeusRun:
failed: dict[str, tuple[str, str]]
def open_ceus_run(run_dir: Path) -> CeusRun:
"""核对运行目录(存在、不是 `.partial`、有 failed.csv 和快照),读 failed.csv。"""
def open_ceus_run(run_dir: Path, delivery_dir: Path, repo_root: Path) -> CeusRun:
"""核对运行目录(存在、不是 `.partial`、有 failed.csv 和快照),读 failed.csv。
快照里 `ceus.yaml` 的 `paths.delivery_dir`(相对路径相对 `repo_root`,同 ceus_quant)与这次
build 的 `delivery_dir` 规范化后不是同一路径,说明运行不是在这份交付包上跑的,报错停下。
"""
if run_dir.name.endswith(".partial"):
raise ValueError(f"CEUS 运行目录是没跑完的 .partial:{run_dir}")
if not run_dir.is_dir():
raise FileNotFoundError(f"CEUS 运行目录不存在:{run_dir}")
if not (run_dir / "snapshot" / "snapshot.json").is_file():
raise FileNotFoundError(f"CEUS 运行目录缺快照 snapshot/snapshot.json:{run_dir}")
ceus_yaml = run_dir / "snapshot" / "ceus.yaml"
ran_on = _typed(
yaml.safe_load(ceus_yaml.read_text(encoding="utf-8"))["paths"]["delivery_dir"], str
)
if (repo_root / ran_on).resolve() != delivery_dir.resolve():
raise ValueError(
f"CEUS 运行不是在这份交付包上跑的:{ceus_yaml} 的 paths.delivery_dir 是 {ran_on},"
f"build 的 delivery_dir 是 {delivery_dir}"
)
with (run_dir / "failed.csv").open(encoding="utf-8", newline="") as handle:
rows = list(csv.reader(handle))
if not rows or rows[0] != _FAILED_HEADER:
@@ -142,6 +159,16 @@ def case_ceus(run: CeusRun, case: CaseRecord) -> LibraryCeus | SkipRecord:
summary = json.loads((case_dir / "summary.json").read_text(encoding="utf-8"))
if summary["video_id"] != case.video_id:
raise ValueError(f"summary.json 的视频号与所在目录不一致:{case_dir}")
if _typed(summary["video_sha256"], str) != hashlib.sha256(case.video.read_bytes()).hexdigest():
raise ValueError(
f"CEUS 运行不是在这份交付包上跑的:{case.video} 的 SHA-256 与 "
f"{case_dir / 'summary.json'} 的 video_sha256 不同"
)
if _typed(summary["injection_offset_s"], float) != case.bolus_start_s:
raise ValueError(
f"CEUS 运行不是在这份交付包上跑的:交付包的推注时间 {case.bolus_start_s} 秒与 "
f"{case_dir / 'summary.json'} 的 injection_offset_s 不同"
)
tic = tic_view(summary, case.video_id)
located = [slot for slot in _SLOTS if tic["keyframes"][slot]["located"]]
images = sorted(p.stem for p in (case_dir / "keyframes").iterdir())
+2 -2
View File
@@ -67,7 +67,6 @@ from cadx.ceus._geometry import (
from cadx.ceus._green_prompt import (
ColorPrompt,
PromptObservation,
collect_prompt_observations,
extract_color_prompt,
prompt_color_mask,
)
@@ -111,6 +110,7 @@ from cadx.ceus._split_screen import (
SplitDetectionError,
detect_split_x,
sample_frames_for_split,
sample_indices,
split_views,
view_rect,
)
@@ -207,7 +207,6 @@ __all__ = [
"align_polygon_to_previous",
"apply_dark_morphology",
"apply_prompt_frame_position_constraints",
"collect_prompt_observations",
"color_marks",
"compute_lognormal_perfusion_metrics",
"compute_tic_for_video",
@@ -255,6 +254,7 @@ __all__ = [
"render_tic_png",
"resample_closed_contour",
"sample_frames_for_split",
"sample_indices",
"score_dark_candidate",
"select_prompt_frames",
"select_reference_region",
+15 -19
View File
@@ -1,19 +1,19 @@
"""第 2 步:判断造影画面在哪一半,并在那一半上选出提示帧。
哪一半是造影画面,各例不一定相同。绿框只画在造影那一半,所以两半各自按同样的办法
(`collect_prompt_observations` + `select_prompt_frames`,参数相同)选提示帧:恰好一半选出
(逐帧 `extract_color_prompt` + `select_prompt_frames`,参数相同)选提示帧:恰好一半选出
至少一帧,这一半就是造影画面;两半都选出或都选不出,这一例失败(`ContrastSideError`)。
旧代码按配置写死造影在左半,造影在右半的病例就找不到绿框。
"""
from collections.abc import Sequence
from collections.abc import Iterable
from dataclasses import dataclass
from typing import Literal
from cadx.ceus._config import KeyframesConfig, PromptSourceConfig
from cadx.ceus._errors import CeusCaseError
from cadx.ceus._geometry import Frame, ViewRect, crop_frame
from cadx.ceus._green_prompt import collect_prompt_observations
from cadx.ceus._green_prompt import PromptObservation, extract_color_prompt
from cadx.ceus._prompt_frames import PromptFrame, select_prompt_frames
PROMPT_STEP = 2
@@ -42,28 +42,24 @@ class ContrastSide:
prompt_frames: list[PromptFrame]
def _prompt_frames_in(
frames: Sequence[Frame],
view: ViewRect,
prompt_cfg: PromptSourceConfig,
keyframes_cfg: KeyframesConfig,
) -> list[PromptFrame]:
observations = collect_prompt_observations(
(crop_frame(frame, view) for frame in frames), prompt_cfg
)
return select_prompt_frames(observations, keyframes_cfg)
def locate_contrast_side(
frames: Sequence[Frame],
frames: Iterable[Frame],
left_view: ViewRect,
right_view: ViewRect,
prompt_cfg: PromptSourceConfig,
keyframes_cfg: KeyframesConfig,
) -> ContrastSide:
"""两半各自选提示帧,恰好一半有就定为造影画面;否则抛 `ContrastSideError`。"""
left = _prompt_frames_in(frames, left_view, prompt_cfg, keyframes_cfg)
right = _prompt_frames_in(frames, right_view, prompt_cfg, keyframes_cfg)
"""两半各自选提示帧,恰好一半有就定为造影画面;否则抛 `ContrastSideError`。
`frames` 只遍历一遍,逐帧在两半上各找一次绿框,可以直接传逐帧解码的生成器。
"""
observations: tuple[list[PromptObservation], list[PromptObservation]] = ([], [])
for index, frame in enumerate(frames):
for found, view in zip(observations, (left_view, right_view), strict=True):
prompt = extract_color_prompt(crop_frame(frame, view), prompt_cfg)
found.append(PromptObservation(frame_index=index, prompt=prompt))
left = select_prompt_frames(observations[0], keyframes_cfg)
right = select_prompt_frames(observations[1], keyframes_cfg)
if left and not right:
return ContrastSide("left", left_view, right_view, left)
if right and not left:
+3 -14
View File
@@ -3,21 +3,20 @@
- `prompt_color_mask`:TIC 采样时要从亮度统计里剔除医生画上去的线,这里给出那组颜色的掩膜,
受 `exclude_prompt_color` 开关和 `prompt_exclude_dilate` 膨胀核控制;
- `extract_color_prompt`:HSV 范围内的像素 → 抹掉顶部文字带 → 形态学闭运算加膨胀把断开的描边
连起来 → 外轮廓 → 按外接框面积 w×h 过滤、取最大者 → 实心填充成掩膜;没有候选返回 None;
- `collect_prompt_observations`:逐帧调上一个函数,没检出的帧用 None 占位,帧号与下标对齐。
连起来 → 外轮廓 → 按外接框面积 w×h 过滤、取最大者 → 实心填充成掩膜;没有候选返回 None。
颜色来自参数:喂绿框那组配置检绿框,喂紫圈那组配置(`PanSourceConfig`)检紫圈。帧是已经切出的
一半画面,返回的外接框和掩膜都是这一半的局部坐标。
"""
from collections.abc import Iterable, Sequence
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Protocol
import cv2
import numpy as np
from cadx.ceus._config import PromptSourceConfig, TicAnalysisConfig
from cadx.ceus._config import TicAnalysisConfig
from cadx.ceus._geometry import AnyArray, Frame, Mask, rect_from_values
# 形态学结构元:5×5,闭运算与膨胀各一次。
@@ -174,13 +173,3 @@ def extract_color_prompt(
contour_area=best.contour_area,
candidate_count=candidate_count,
)
def collect_prompt_observations(
source_frames: Iterable[Frame], prompt_cfg: PromptSourceConfig
) -> list[PromptObservation]:
"""逐帧找绿框,交出与帧序列等长的观测列表;没检出的帧 `prompt` 为 None。"""
return [
PromptObservation(frame_index=index, prompt=extract_color_prompt(frame, prompt_cfg))
for index, frame in enumerate(source_frames)
]
+46 -17
View File
@@ -1,8 +1,8 @@
"""ceus 对外的两段接口,中间由 `scripts/ceus_quant.py` 调 segment 做第 3 步分割。
- `prepare_segmentation`(第 1–2 步):读视频、找分屏线、配准两半、判断造影画面在哪一半、选提示帧,
交出要交给分割的那一半(灰阶那一半,`target_view`)的逐帧图像和提示帧掩膜(`target_view`
局部坐标)。
交出提示帧掩膜(`target_view` 局部坐标);要交给分割的那一半(灰阶那一半,`target_view`)的
逐帧图像由 `target_frames()` 重新解码交出。
- `quantify_case`(第 4–9 步):拿分割得到的逐帧掩膜(`target_view` 局部坐标)整理成平滑轮廓并与
绿框融合、挪回造影那一半,找暗区、选背景圆、逐帧采 TIC、拟合、定位 7 个关键帧、算灌注指标。
@@ -47,7 +47,7 @@ from cadx.ceus._polygons import (
)
from cadx.ceus._reference_region import select_reference_region
from cadx.ceus._registration import HalfRegistration, register_halves
from cadx.ceus._split_screen import split_views
from cadx.ceus._split_screen import sample_indices, split_views
from cadx.ceus._tic import CURVE_NAMES, TicResult, compute_tic_for_video
from cadx.ceus._tic_curves import TicCurves, from_tic_result
from cadx.ceus._tic_fitting import LognormalFitResult, evaluate_lognormal
@@ -86,10 +86,10 @@ class SegmentationInput:
`offset`:造影那一半局部坐标 → `target_view` 局部坐标的平移 `(dx, dy)`,取互相关的估计;
- `contrast`:哪一半是造影画面、两个视图矩形(整帧坐标)、提示帧(造影那一半局部坐标的
绿框掩膜);
- `target_frames`:要交给分割的那一半(`contrast.target_view`,灰阶画面)的逐帧 BGR 图像,
与帧号一一对应;
- `prompt_masks`:提示帧帧号 → 这一帧绿框挪到 `target_view` 局部坐标后的布尔掩膜;
- `first_contrast_frame`:造影那一半的第一帧(选背景圆用)。
要交给分割的那一半画面不存在这里,由 `target_frames()` 重新解码逐帧交出。
"""
video_path: Path
@@ -101,10 +101,14 @@ class SegmentationInput:
registration: HalfRegistration
offset: Offset
contrast: ContrastSide
target_frames: list[Frame]
prompt_masks: dict[int, Mask]
first_contrast_frame: Frame
def target_frames(self) -> Iterator[Frame]:
"""重新解码,逐帧交出分割那一半(`contrast.target_view`,灰阶画面)的 BGR 图像。"""
for _, frame in _checked_frames(VideoFile(self.video_path), self.frame_count):
yield crop_frame(frame, self.contrast.target_view)
@dataclass(frozen=True)
class QuantResult:
@@ -129,9 +133,25 @@ class QuantResult:
pan_detected_fraction: float
def _checked_frames(video: VideoFile, frame_count: int) -> Iterator[tuple[int, Frame]]:
"""逐帧解码,遍历完核对解码帧数与容器记录的一致。"""
decoded = 0
for index, frame in video.iter_frames():
decoded += 1
yield index, frame
if decoded != frame_count:
raise VideoDecodeError(
f"解码帧数 {decoded} 与容器记录的帧数 {frame_count} 不一致:{video.path}"
)
def prepare_segmentation(video_path: Path, cfg: CeusConfig) -> SegmentationInput:
"""第 1–2 步:解码、分屏、配准两半、找绿框定造影那一半、选提示帧,交出分割的输入。
不把整段视频留在内存里,逐帧解码两遍:第一遍只留找分屏线和配准用的抽样帧;第二遍逐帧在两半
上找绿框。抽样帧号与 `split_views`、`register_halves` 自己抽的相同,所以结果与把整段帧交给
它们一样。分割那一半的画面由 `SegmentationInput.target_frames()` 到用时再解码。
视频读不了或读出的不合规抛 `VideoDecodeError`、找不到可信分屏线抛 `SplitDetectionError`、两半
配准失败(含两种方法不一致)抛 `RegistrationError`(都是第 1 步),两半都有或都没有绿框抛
`ContrastSideError`(第 2 步),都是 `CeusCaseError`。
@@ -139,22 +159,32 @@ def prepare_segmentation(video_path: Path, cfg: CeusConfig) -> SegmentationInput
video = VideoFile(video_path)
fps, frame_count, width, height = video.probe()
frame_times_s = video.frame_times_s()
full_frames = [frame for _, frame in video.iter_frames()]
if len(full_frames) != frame_count:
raise VideoDecodeError(
f"解码帧数 {len(full_frames)} 与容器记录的帧数 {frame_count} 不一致:{video_path}"
)
split_x, left_view, right_view = split_views(full_frames, cfg.views)
split_at = set(sample_indices(frame_count, cfg.views.split_sample_frames))
register_at = set(sample_indices(frame_count, cfg.registration.sample_frames))
split_frames: list[Frame] = []
register_frames: list[Frame] = []
for index, frame in _checked_frames(video, frame_count):
if index in split_at:
split_frames.append(frame)
if index in register_at:
register_frames.append(frame)
split_x, left_view, right_view = split_views(split_frames, cfg.views)
registration = register_halves(
full_frames, left_view, right_view, cfg.registration, cfg.prompt_source, cfg.pan_source
register_frames, left_view, right_view, cfg.registration, cfg.prompt_source, cfg.pan_source
)
first_frame = split_frames[0] # 抽样帧号从 0 起,第 0 帧总在里面
del split_frames, register_frames
contrast = locate_contrast_side(
full_frames, left_view, right_view, cfg.prompt_source, cfg.keyframes
(frame for _, frame in _checked_frames(video, frame_count)),
left_view,
right_view,
cfg.prompt_source,
cfg.keyframes,
)
offset = registration.contrast_to_target(contrast.side)
target_frames = [crop_frame(frame, contrast.target_view).copy() for frame in full_frames]
first_contrast_frame = crop_frame(full_frames[0], contrast.contrast_view).copy()
first_contrast_frame = crop_frame(first_frame, contrast.contrast_view).copy()
prompt_masks = {
prompt.frame_index: map_source_mask_to_target(prompt.mask, offset, contrast.target_view)
for prompt in contrast.prompt_frames
@@ -169,7 +199,6 @@ def prepare_segmentation(video_path: Path, cfg: CeusConfig) -> SegmentationInput
registration=registration,
offset=offset,
contrast=contrast,
target_frames=target_frames,
prompt_masks=prompt_masks,
first_contrast_frame=first_contrast_frame,
)
+10 -6
View File
@@ -162,15 +162,19 @@ def view_rect(
return rect
def sample_indices(frame_count: int, sample_count: int) -> list[int]:
"""沿时间轴均匀抽 `sample_count` 个帧号(去重、升序);帧数不超过它时全用。"""
if sample_count <= 0 or frame_count <= sample_count:
return list(range(frame_count))
indices = np.linspace(0, frame_count - 1, num=sample_count, dtype=np.int32)
return sorted({int(index) for index in indices.tolist()})
def sample_frames_for_split(frames: Sequence[Frame], sample_count: int) -> list[Frame]:
"""沿时间轴均匀抽 `sample_count` 帧(下标去重);帧数不超过它时全用。"""
"""按 `sample_indices` 均匀抽帧;已经抽好的帧(不多于 `sample_count`)原样全用。"""
if not frames:
raise SplitDetectionError("帧序列为空,无法找分屏线。")
if sample_count <= 0 or len(frames) <= sample_count:
return list(frames)
indices = np.linspace(0, len(frames) - 1, num=sample_count, dtype=np.int32)
unique_indices = sorted({int(index) for index in indices.tolist()})
return [frames[index] for index in unique_indices]
return [frames[index] for index in sample_indices(len(frames), sample_count)]
def split_views(frames: Sequence[Frame], views_cfg: ViewsConfig) -> tuple[int, ViewRect, ViewRect]:
+8 -2
View File
@@ -1,5 +1,11 @@
"""runs:构建号怎么取、运行快照记什么。会诊运行目录等会诊那一步再加。"""
from cadx.runs._snapshot import GitState, make_build_id, read_git_state, write_snapshot
from cadx.runs._snapshot import (
GitState,
finish_snapshot,
make_build_id,
read_git_state,
write_snapshot,
)
__all__ = ["GitState", "make_build_id", "read_git_state", "write_snapshot"]
__all__ = ["GitState", "finish_snapshot", "make_build_id", "read_git_state", "write_snapshot"]
+20 -3
View File
@@ -1,4 +1,9 @@
"""构建号与运行快照:一次构建用的是哪份代码、哪份配置、哪个环境。"""
"""构建号与运行快照:一次构建用的是哪份代码、哪份配置、哪个环境。
快照分两次写:开跑时 `write_snapshot` 记下代码、配置原文和环境(运行期间改了配置文件也不影响
快照),收尾时 `finish_snapshot` 另写 `finished.json`,只记结束时间和运行结束才知道的值。两次
都只新建文件,不改已写的。
"""
import hashlib
import json
@@ -55,7 +60,6 @@ def write_snapshot(
build_id: str,
git: GitState,
started_at: datetime,
finished_at: datetime,
uv_lock: Path,
configs: Mapping[str, Path],
) -> None:
@@ -73,9 +77,22 @@ def write_snapshot(
"python_version": sys.version,
"uv_lock_sha256": _sha256(uv_lock),
"started_at": started_at.isoformat(),
"finished_at": finished_at.isoformat(),
"config_sha256": copied,
}
(snapshot_dir / "snapshot.json").write_text(
json.dumps(snapshot, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
def finish_snapshot(
snapshot_dir: Path, finished_at: datetime, record: Mapping[str, object]
) -> None:
"""收尾时在快照目录里新建 `finished.json`:结束时间加 `record`(运行结束才知道的值)。
`finished.json` 已存在(同一快照收两次尾)或 `record` 里有 `finished_at` 键都报错。
"""
if "finished_at" in record:
raise ValueError("record 里不能有 finished_at,结束时间由参数 finished_at 给")
content = {"finished_at": finished_at.isoformat(), **record}
with (snapshot_dir / "finished.json").open("x", encoding="utf-8") as handle:
handle.write(json.dumps(content, ensure_ascii=False, indent=2) + "\n")
+121 -54
View File
@@ -4,13 +4,19 @@
`000000.jpg` 起的 jpg 序列(质量 95)写进一个目录交给 `init_state`;每个提示帧用 `add_new_mask`
加进同一个对象;从最早的提示帧向后传播到末帧,再从它向前传播到第 0 帧;logits > 0 为病灶。
现在在同一进程里调用,jpg 临时目录由这里建、用完即删。
`init_state` 读 jpg 时换成 `_load_jpg_frames`:MedSAM2 原来的读法先在 CPU 上建整段 float32 张量
(N×3×512×512,2288 帧约 7.2GB)再整个搬上显卡,这份 CPU 张量是整个流程的内存峰值。换上的读法
直接在存放帧的设备上建张量、逐帧写入,算术与原来逐位相同(MedSAM2 源码不改)。
"""
import inspect
import os
import shutil
import subprocess
import tempfile
from collections.abc import Mapping, Sequence
from collections.abc import Generator, Iterable, Mapping
from contextlib import contextmanager
from pathlib import Path
from typing import Any
@@ -58,33 +64,25 @@ def cuda_peak_memory_bytes(device: str) -> int | None:
return int(torch.cuda.max_memory_allocated(device))
def _check_inputs(
frames: Sequence[Frame], prompt_masks: Mapping[int, Mask], cfg: SegmentConfig
) -> tuple[int, int]:
"""校验输入并返回帧的 (高, 宽)。"""
def _check_files(cfg: SegmentConfig) -> None:
for path in (cfg.medsam2_dir / "sam2" / cfg.sam2_config, cfg.checkpoint):
if not path.is_file():
raise FileNotFoundError(f"MedSAM2 缺少文件:{path}")
if len(frames) == 0:
raise ValueError("没有要分割的帧")
height, width = frames[0].shape[:2]
for index, frame in enumerate(frames):
if frame.dtype != np.uint8 or frame.shape != (height, width, 3):
raise ValueError(
f"第 {index} 帧应为 ({height}, {width}, 3) 的 uint8 BGR 图像,"
f"收到 {frame.shape} {frame.dtype}"
)
def _check_prompts(
prompt_masks: Mapping[int, Mask], frame_count: int, height: int, width: int
) -> None:
if len(prompt_masks) == 0:
raise ValueError("至少需要一个提示帧")
for index, mask in prompt_masks.items():
if not 0 <= index < len(frames):
raise ValueError(f"提示帧帧号 {index} 不在 [0, {len(frames)}) 内")
if not 0 <= index < frame_count:
raise ValueError(f"提示帧帧号 {index} 不在 [0, {frame_count}) 内")
if mask.dtype != np.bool_ or mask.shape != (height, width):
raise ValueError(
f"提示帧 {index} 的掩码应为 ({height}, {width}) 的布尔数组,"
f"收到 {mask.shape} {mask.dtype}"
)
return height, width
def _check_device(device: str) -> None:
@@ -110,70 +108,139 @@ def _check_sam2_source(medsam2_dir: Path) -> None:
raise RuntimeError(f"环境里的 sam2 来自 {loaded},与配置的 {expected} 不一致")
def _write_jpgs(frames: Sequence[Frame], frames_dir: Path) -> None:
def _write_jpgs(frames: Iterable[Frame], frames_dir: Path) -> tuple[int, int, int]:
"""逐帧写 jpg(帧不留在内存里),返回(帧数, 高, 宽);各帧须是同一尺寸的 uint8 BGR 图像。"""
count, shape = 0, None
for index, frame in enumerate(frames):
shape = frame.shape if shape is None else shape
if frame.dtype != np.uint8 or len(shape) != 3 or frame.shape != shape or shape[2] != 3:
raise ValueError(
f"第 {index} 帧应为与第 0 帧同尺寸 {shape} 的 uint8 BGR 图像,"
f"收到 {frame.shape} {frame.dtype}"
)
path = frames_dir / f"{index:06d}.jpg"
if not cv2.imwrite(str(path), frame, [cv2.IMWRITE_JPEG_QUALITY, _JPEG_QUALITY]):
raise RuntimeError(f"写分割用的 jpg 失败:{path}")
count += 1
if shape is None:
raise ValueError("没有要分割的帧")
return count, shape[0], shape[1]
def _load_jpg_frames(
*,
video_path: str,
image_size: int,
offload_video_to_cpu: bool,
async_loading_frames: bool,
compute_device: torch.device,
) -> tuple[torch.Tensor, int, int]:
"""代替 `sam2.utils.misc.load_video_frames` 读 `_write_jpgs` 写的目录,参数和返回值与它相同。
照原函数 `load_video_frames_from_jpg_images`:同一个 `_load_img_as_tensor`(float64)逐帧写进
float32 张量,再减均值、除标准差(均值、标准差取原函数的参数默认值)。不同只在张量直接建在
存放设备上(`offload_video_to_cpu` 为否时是显卡),不先在 CPU 上建整段再搬:float64 转 float32
是同一种舍入,归一化原来也在这个设备上做,结果逐位相同。
"""
from sam2.utils.misc import ( # pyright: ignore[reportMissingTypeStubs]
_load_img_as_tensor, # pyright: ignore[reportPrivateUsage, reportUnknownVariableType]
load_video_frames_from_jpg_images, # pyright: ignore[reportUnknownVariableType]
)
if async_loading_frames:
raise ValueError("_load_jpg_frames 不支持 async_loading_frames")
defaults = inspect.signature(load_video_frames_from_jpg_images).parameters # pyright: ignore[reportUnknownArgumentType]
device = torch.device("cpu") if offload_video_to_cpu else compute_device
img_mean = torch.tensor(defaults["img_mean"].default, dtype=torch.float32)[:, None, None]
img_std = torch.tensor(defaults["img_std"].default, dtype=torch.float32)[:, None, None]
names = sorted(name for name in os.listdir(video_path) if name.endswith(".jpg"))
if not names:
raise RuntimeError(f"分割用的 jpg 目录是空的:{video_path}")
images = torch.zeros(len(names), 3, image_size, image_size, dtype=torch.float32, device=device)
height = width = 0
for index, name in enumerate(names):
path = os.path.join(video_path, name)
images[index], height, width = _load_img_as_tensor(path, image_size) # pyright: ignore[reportUnknownVariableType]
images -= img_mean.to(device)
images /= img_std.to(device)
return images, height, width
@contextmanager
def _jpg_frames_on_device() -> Generator[None]:
"""`init_state` 期间把 sam2 视频预测器读帧的函数换成 `_load_jpg_frames`,退出时换回。"""
import sam2.sam2_video_predictor as video_predictor # pyright: ignore[reportMissingTypeStubs]
original = video_predictor.load_video_frames # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
video_predictor.load_video_frames = _load_jpg_frames
try:
yield
finally:
video_predictor.load_video_frames = original
def _to_mask(mask_logits: torch.Tensor) -> Mask:
return np.squeeze((mask_logits > 0.0).detach().cpu().numpy()).astype(np.bool_)
def _propagate(predictor: Any, state: Any, first_prompt: int) -> dict[int, Mask]:
"""从最早的提示帧向后传播到末帧,再向前传播到第 0 帧。"""
propagated: dict[int, Mask] = {}
def _propagate(predictor: Any, state: Any, first_prompt: int, masks: Mask) -> Mask:
"""从最早的提示帧向后传播到末帧,再向前传播到第 0 帧,逐帧写进 `masks`(后写的覆盖先写的)。
返回哪些帧写过。
"""
written = np.zeros(masks.shape[0], dtype=np.bool_)
passes = (
{"start_frame_idx": first_prompt},
{
"start_frame_idx": first_prompt,
"max_frame_num_to_track": first_prompt + 1,
"reverse": True,
},
)
with torch.inference_mode():
for frame_idx, obj_ids, mask_logits in predictor.propagate_in_video(
state, start_frame_idx=first_prompt
):
propagated[int(frame_idx)] = _to_mask(mask_logits[list(obj_ids).index(_OBJECT_ID)])
for frame_idx, obj_ids, mask_logits in predictor.propagate_in_video(
state,
start_frame_idx=first_prompt,
max_frame_num_to_track=first_prompt + 1,
reverse=True,
):
propagated[int(frame_idx)] = _to_mask(mask_logits[list(obj_ids).index(_OBJECT_ID)])
return propagated
for kwargs in passes:
for frame_idx, obj_ids, mask_logits in predictor.propagate_in_video(state, **kwargs):
mask = _to_mask(mask_logits[list(obj_ids).index(_OBJECT_ID)])
if mask.shape != masks.shape[1:]:
raise RuntimeError(f"MedSAM2 输出掩码形状 {mask.shape},应为 {masks.shape[1:]}")
masks[int(frame_idx)] = mask
written[int(frame_idx)] = True
return written
def segment_video(
frames: Sequence[Frame], prompt_masks: Mapping[int, Mask], cfg: SegmentConfig
frames: Iterable[Frame], prompt_masks: Mapping[int, Mask], cfg: SegmentConfig
) -> Mask:
"""把提示帧的病灶掩码传播到每一帧。
`frames` 是逐帧 BGR 图像(同一尺寸),`prompt_masks` 是提示帧帧号 → 与帧同尺寸的布尔掩码。
返回 `(帧数, 高, 宽)` 的布尔数组,与输入帧一一对应。
`frames` 是逐帧 BGR 图像(同一尺寸),只遍历一遍、边读边写成 jpg,可以直接传逐帧解码的生成器;
`prompt_masks` 是提示帧帧号 → 与帧同尺寸的布尔掩码。返回 `(帧数, 高, 宽)` 的布尔数组,与输入帧
一一对应。
"""
height, width = _check_inputs(frames, prompt_masks, cfg)
_check_files(cfg)
_check_device(cfg.device)
_check_sam2_source(cfg.medsam2_dir)
# sam2 没有类型标注,参数类型是 Unknown;返回的 predictor 按 Any 用。
from sam2.build_sam import build_sam2_video_predictor # pyright: ignore[reportUnknownVariableType]
predictor: Any = build_sam2_video_predictor(
config_file=cfg.sam2_config, ckpt_path=str(cfg.checkpoint), device=cfg.device
)
with tempfile.TemporaryDirectory(prefix="cadx-medsam2-") as tmp:
frames_dir = Path(tmp)
_write_jpgs(frames, frames_dir)
state: Any = predictor.init_state(
str(frames_dir),
offload_video_to_cpu=cfg.offload_video_to_cpu,
offload_state_to_cpu=cfg.offload_state_to_cpu,
frame_count, height, width = _write_jpgs(frames, frames_dir)
_check_prompts(prompt_masks, frame_count, height, width)
predictor: Any = build_sam2_video_predictor(
config_file=cfg.sam2_config, ckpt_path=str(cfg.checkpoint), device=cfg.device
)
with _jpg_frames_on_device():
state: Any = predictor.init_state(
str(frames_dir),
offload_video_to_cpu=cfg.offload_video_to_cpu,
offload_state_to_cpu=cfg.offload_state_to_cpu,
)
for index in sorted(prompt_masks):
predictor.add_new_mask(state, frame_idx=index, obj_id=_OBJECT_ID, mask=prompt_masks[index])
propagated = _propagate(predictor, state, min(prompt_masks))
missing = [index for index in range(len(frames)) if index not in propagated]
masks = np.zeros((frame_count, height, width), dtype=np.bool_)
written = _propagate(predictor, state, min(prompt_masks), masks)
missing = np.flatnonzero(~written).tolist()
if missing:
raise RuntimeError(f"MedSAM2 传播后缺 {len(missing)} 帧的掩码,前几帧:{missing[:5]}")
stack = np.stack([propagated[index] for index in range(len(frames))], axis=0)
if stack.shape != (len(frames), height, width):
raise RuntimeError(
f"MedSAM2 输出掩码形状 {stack.shape},应为 {(len(frames), height, width)}"
)
return stack
return masks