"""ceus 对外两段接口的端到端测试：合成一段分屏造影视频，第二段喂合成掩膜（不调分割模型）。 视频 400×200、10 fps、40 秒（400 帧），分屏线在 x=200（这一列全黑）。造影画面在右半：病灶是整帧 [250, 369] × [60, 159] 的方块，医生的绿框是它的空心描边；病灶内部亮度按 Lognormal 团注曲线从 基线 40 升到约 170 再回落，内部另有一块一直发黑的囊（暗区）；每帧加一点随机噪声，免得相邻帧 被当成重复帧。两半各有一块位置相同、带纹理、随时间轻微起伏的扇形区（`fake_split_scene`， 配准要靠它），左半是灰阶画面，右半（造影）调暗。 手算的预期： - 绿框在右半 → 造影画面是右半，分割那一半是左半 [0, 200)； - 绿框挪到左半局部坐标：x = 250 − 200 = 50 到 369 − 200 = 169； - 推注时间 X = 5 秒 → F1 在视频第 20 − 5 = 15 秒（第 150 帧），F2 在第 60 − 5 = 55 秒， 超过 40 秒的视频末尾，未定位。 """ import json import math import numpy as np import pytest from cadx.ceus import ( ANCHOR_SLOTS, CeusConfig, NoValidContourError, SegmentationInput, VideoFile, crop_frame, keyframe_images, locate_contrast_side, pack_tic_curves, pack_trajectory, prepare_segmentation, quant_summary, quantify_case, register_halves, render_case_overlay, render_tic_png, split_views, tic_plot_markers, unpack_tic_curves, unpack_trajectory, ) from fake_ceus_video import BOLUS, FPS, FRAMES, HEIGHT, LESION, SPLIT, WIDTH, write_video _INJECTION_OFFSET_S = 5.0 @pytest.fixture(scope="module") def config() -> CeusConfig: from conftest import load_ceus_section return CeusConfig.model_validate(load_ceus_section()) @pytest.fixture(scope="module") def prepared(tmp_path_factory: pytest.TempPathFactory, config: CeusConfig) -> SegmentationInput: path = tmp_path_factory.mktemp("ceus") / "synthetic.mp4" write_video(path, green_box=True) return prepare_segmentation(path, config) def _lesion_masks(prepared: SegmentationInput) -> np.ndarray: """合成「分割结果」：每帧都是病灶方块挪到左半局部坐标后的位置（略往里收 3 像素）。""" view = prepared.contrast.target_view masks = np.zeros((prepared.frame_count, view.height, view.width), dtype=bool) x1, y1, x2, y2 = LESION masks[:, y1 + 3 : y2 - 2, x1 - SPLIT + 3 : x2 - SPLIT - 2] = True return masks def test_prepare_segmentation_finds_contrast_half_and_maps_prompts( prepared: SegmentationInput, ) -> None: assert prepared.frame_count == FRAMES assert prepared.frame_size == (WIDTH, HEIGHT) assert prepared.contrast.side == "right" # 两半扇形对齐：两种配准都估出 (0, 0)；造影在右半，平移取反后仍是 (0, 0)。 assert prepared.registration.correlation == (0, 0) assert prepared.registration.disagreement_px <= 1 assert prepared.offset == (0, 0) target = prepared.contrast.target_view assert (target.x1, target.x2) == (0, prepared.split_x) assert abs(prepared.split_x - SPLIT) <= 1 target_frames = list(prepared.target_frames()) assert len(target_frames) == FRAMES assert all(frame.shape == (target.height, target.width, 3) for frame in target_frames) assert sorted(prepared.prompt_masks) == [p.frame_index for p in prepared.contrast.prompt_frames] assert 0 in prepared.prompt_masks xs = np.where(prepared.prompt_masks[0])[1] # 描边 3 像素宽、找绿框时再膨胀 2 像素、mp4 压缩会糊开边缘，容差放到 5 像素。 assert abs(int(xs.min()) - 50) <= 5 assert abs(int(xs.max()) - 169) <= 5 def test_streaming_prepare_matches_whole_video_in_memory( prepared: SegmentationInput, config: CeusConfig ) -> None: """逐帧解码的第 1–2 步与「整段帧读进内存再交给各步」逐像素相同（改成流式之前就是后一种）。""" frames = [frame for _, frame in VideoFile(prepared.video_path).iter_frames()] split_x, left, right = split_views(frames, config.views) registration = register_halves( frames, left, right, config.registration, config.prompt_source, config.pan_source ) contrast = locate_contrast_side(frames, left, right, config.prompt_source, config.keyframes) assert prepared.split_x == split_x assert prepared.registration == registration assert prepared.contrast.side == contrast.side assert (prepared.contrast.contrast_view, prepared.contrast.target_view) == ( contrast.contrast_view, contrast.target_view, ) assert [p.frame_index for p in prepared.contrast.prompt_frames] == [ p.frame_index for p in contrast.prompt_frames ] for ours, whole in zip(prepared.contrast.prompt_frames, contrast.prompt_frames, strict=True): assert np.array_equal(ours.mask, whole.mask) target_frames = list(prepared.target_frames()) assert len(target_frames) == len(frames) for ours, frame in zip(target_frames, frames, strict=True): assert np.array_equal(ours, crop_frame(frame, contrast.target_view)) assert np.array_equal( prepared.first_contrast_frame, crop_frame(frames[0], contrast.contrast_view) ) def test_quantify_case_end_to_end(prepared: SegmentationInput, config: CeusConfig) -> None: result = quantify_case(prepared, _lesion_masks(prepared), _INJECTION_OFFSET_S, config) assert result.trajectory.source_polygons.shape == (FRAMES, 64, 2) assert result.lesion_fit_ready is True assert list(result.anchors) == list(ANCHOR_SLOTS) for slot in ("D1", "D2", "D3"): assert result.anchors[slot].status == "ok", result.anchors[slot].reason assert result.anchors["F1"].frame_index == 150 assert result.anchors["F1"].frame_time_video_s == pytest.approx(15.0, abs=0.06) assert result.anchors["F2"].status == "not_covered" assert result.anchors["F2"].reason is not None assert result.perfusion["roi_perf"] is not None peak_time = BOLUS.arrival_offset_s + math.exp(BOLUS.mu - BOLUS.sigma**2) assert result.perfusion["roi_perf"].ttp_s == pytest.approx(peak_time, abs=0.5) # 画面上没有紫圈。 assert result.pan_detected_fraction == 0.0 assert result.tic.reference_valid is False # 暗区（囊）被找出来，去暗区曲线比病灶整体亮。 peak_row = result.tic.rows[round(peak_time * FPS)] assert peak_row["dark_area_px"] > 0 assert peak_row["roi_perf_raw"] > peak_row["roi_all_raw"] restored = unpack_trajectory(pack_trajectory(result.trajectory)) assert restored.calibration["contrast_side"] == "right" curves = unpack_tic_curves(pack_tic_curves(result.curves)) assert curves.injection_offset_s == pytest.approx(_INJECTION_OFFSET_S) summary = quant_summary(result) json.dumps(summary, allow_nan=False) assert summary["anchor_frames"]["F1"]["frame_index"] == 150 assert summary["perfusion_metrics"]["roi_perf"] is not None markers = tic_plot_markers(result.anchors) assert [marker.label.split()[0] for marker in markers][:3] == ["D1", "D2", "D3"] assert render_tic_png(result.tic, markers=markers).startswith(b"\x89PNG") assert len(render_case_overlay(prepared, result)) > 0 images = keyframe_images(prepared, result.anchors) assert "F1" in images and "F2" not in images assert all(image.shape == (HEIGHT, WIDTH, 3) for image in images.values()) def test_empty_segmentation_fails_case_at_step_4( prepared: SegmentationInput, config: CeusConfig ) -> None: empty = np.zeros_like(_lesion_masks(prepared)) with pytest.raises(NoValidContourError) as raised: quantify_case(prepared, empty, _INJECTION_OFFSET_S, config) assert raised.value.step == 4 def test_mask_shape_mismatch_is_rejected(prepared: SegmentationInput, config: CeusConfig) -> None: masks = _lesion_masks(prepared)[:-1] with pytest.raises(ValueError, match="分割掩膜"): quantify_case(prepared, masks, _INJECTION_OFFSET_S, config)