feat(ceus): 任务书 02 第一阶段——底层和分割之前的算法;修复 C14

任务书 01 遗留修复(C14):
- library 新增 check_key_image(path):关键图能逐像素保真重编码时返回 None,否则返回原因;
  _write_clean_png 的判断移过去,自身保留断言性质报错(走到这里说明 build 漏了检查)。
- build 写库前对每例调用,返回原因的整例跳过,记 C14「关键图能原样重编码」,涉及值写原因。
- 原来三条「关键图格式不对整次构建停下」的测试改为整例跳过,另加非 PNG、library 漏检报错两条。
- 结构.md 第八部分、guides/装配两个库.md 相应几行改掉。

第一阶段(src/cadx/ceus/):
- 照搬旧 geometry、video_decoder、research_config、split_screen、green_prompt、prompt_frames、
  mapping,去掉产品层,注释改中文。
- 配置类去掉全部默认值,strict、extra=forbid、frozen;新建 config/ceus.yaml(paths、ceus 两段),
  数值照抄旧 yaml,逐项注明出处。删去 source_view/target_view、effective_region、mapping 锚点;
  旧代码写死的分屏抽帧数 25 和搜索区间 (0.35, 0.65) 改成配置,值不变。
- 视频解码改为按路径用 cv2 逐帧读。
- 坐标换算改为两个视图矩形左上角之差,不再用按某种宽度写死的锚点。
- 造影画面在哪一半:两半各自选提示帧,恰好一半有才定,否则抛 ContrastSideError(CeusCaseError
  子类,带步号和原因,供第三阶段记 failed.csv)。
- 旧测试照搬(改 import 与配置读取方式),因上述改动不再成立的删去;新增两处修改的手算测试。
- import-linter 加「ceus 不引用其他 cadx 模块」。依赖加 numpy、opencv-python-headless。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-16 03:46:56 -04:00
parent de1d6b2d51
commit 9a916c65a5
29 changed files with 3634 additions and 45 deletions
+40 -9
View File
@@ -4,6 +4,7 @@
"""
import hashlib
import json
import random
import string
from collections import Counter
@@ -20,10 +21,11 @@ from cadx.delivery import (
SKIPPED_NOTE,
CaseRecord,
CodeMaps,
SkipRecord,
read_delivery,
review_sentences,
)
from cadx.library import LibraryCase, write_library
from cadx.library import LibraryCase, check_key_image, write_library
from cadx.runs import make_build_id, read_git_state, write_snapshot
ALIAS_ALPHABET = string.ascii_lowercase + string.digits
@@ -31,6 +33,9 @@ ALIAS_LENGTH = 4
# 化名里出现亚型缩写会被泄露扫描(不分大小写)误报,抽到就重抽。
_ALIAS_BANNED = ("ipmn", "spn", "mcn", "scn")
C14 = "C14"
C14_DESCRIPTION = "关键图能原样重编码"
@dataclass(frozen=True)
class BuildSummary:
@@ -75,6 +80,30 @@ def _clinical_fields(
}
def _check_key_images(
cases: Sequence[CaseRecord], delivery_dir: Path
) -> tuple[list[CaseRecord], list[SkipRecord]]:
"""C14:关键图不能逐像素保真重编码的病例整例跳过,原因写进涉及值。"""
kept: list[CaseRecord] = []
skipped: list[SkipRecord] = []
for case in cases:
reason = check_key_image(case.key_image)
if reason is None:
kept.append(case)
continue
skipped.append(
SkipRecord(
case=case.case_no,
check=C14,
description=C14_DESCRIPTION,
files=(case.key_image.relative_to(delivery_dir).as_posix(),),
columns=(),
value_json=json.dumps(reason, ensure_ascii=False),
)
)
return kept, skipped
def _sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
@@ -91,6 +120,8 @@ def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary
config = load_build_config(config_path, repo_root)
code_maps = load_code_maps(config.code_maps) if config.code_maps is not None else None
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]
git = read_git_state(repo_root)
build_id = make_build_id(now, git)
@@ -104,8 +135,8 @@ def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary
for partial in (library_partial, answers_partial):
partial.mkdir(parents=True, exist_ok=False)
aliases = make_aliases([case.case_no for case in delivery.cases], config.alias_seed)
by_alias = sorted(delivery.cases, key=lambda case: aliases[case.case_no])
aliases = make_aliases([case.case_no for case in passed], config.alias_seed)
by_alias = sorted(passed, key=lambda case: aliases[case.case_no])
configs = {"build.yaml": config_path}
if config.code_maps is not None:
@@ -152,7 +183,7 @@ def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary
skipped_note=SKIPPED_NOTE,
aliases=[
AliasRow(alias=aliases[c.case_no], case_no=c.case_no, video_id=c.video_id)
for c in delivery.cases
for c in passed
],
diagnoses=[
DiagnosisRow(
@@ -163,7 +194,7 @@ def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary
mr_code=c.diagnosis.mr,
name=None if code_maps is None else code_maps["Diagnosis"][c.diagnosis.clinical],
)
for c in delivery.cases
for c in passed
],
skipped=[
SkippedRow(
@@ -174,7 +205,7 @@ def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary
columns=s.columns,
value_json=s.value_json,
)
for s in delivery.skipped
for s in skipped
],
reviews=reviews,
)
@@ -183,14 +214,14 @@ def run_build(config_path: Path, repo_root: Path, now: datetime) -> BuildSummary
_rename(answers_partial, answers_dir)
_rename(library_partial, library_dir)
by_check = Counter(s.check for s in delivery.skipped)
by_check = Counter(s.check for s in skipped)
return BuildSummary(
build_id=build_id,
library_dir=library_dir,
answers_dir=answers_dir,
total_cases=delivery.total_cases,
passed_cases=len(delivery.cases),
skipped_cases=len(delivery.skipped) - delivery.unrecognized_rows,
passed_cases=len(passed),
skipped_cases=len(skipped) - delivery.unrecognized_rows,
unrecognized_rows=delivery.unrecognized_rows,
skipped_by_check=dict(sorted(by_check.items())),
review_rows=len(reviews),
+113
View File
@@ -0,0 +1,113 @@
"""ceus:CEUS 定量的全部算法和参数,纯计算,不引用其他 cadx 模块。
目前有第 1、2 步(视频解码、分屏、找绿框、选提示帧、判断造影画面在哪一半)和两半之间的坐标换算;
第 3 步以后随任务书 02 第二阶段加入。
某一例不满足算法前提时抛 `CeusCaseError`(带失败的步号和原因)。
"""
from cadx.ceus._config import (
AnchorArrivalConfig,
AnchorFramesConfig,
AnchorLatePhaseConfig,
CeusConfig,
ConstraintsConfig,
InteriorDarkRegionConfig,
KeyframePositionConstraint,
KeyframesConfig,
PanSourceConfig,
PromptSourceConfig,
ReferenceRegionConfig,
RoiShapeConfig,
SmoothingConfig,
TicAnalysisConfig,
TicBaselineConfig,
TicFittingConfig,
ViewsConfig,
)
from cadx.ceus._contrast_side import ContrastSide, ContrastSideError, locate_contrast_side
from cadx.ceus._errors import CeusCaseError
from cadx.ceus._geometry import (
ViewRect,
crop_frame,
local_rect_within,
mask_iou,
odd_window,
optional_positive_int,
polygon_area,
polygon_center,
polygon_to_mask,
polygons_to_mask,
rect_from_values,
resample_closed_contour,
triangular_smooth,
)
from cadx.ceus._green_prompt import (
ColorPrompt,
PromptObservation,
collect_prompt_observations,
extract_color_prompt,
prompt_color_mask,
)
from cadx.ceus._mapping import map_source_mask_to_target, target_polygons_to_source, view_offset
from cadx.ceus._prompt_frames import PromptFrame, select_prompt_frames
from cadx.ceus._split_screen import (
SplitDetectionError,
detect_split_x,
sample_frames_for_split,
split_views,
view_rect,
)
from cadx.ceus._video import VideoFile
__all__ = [
"AnchorArrivalConfig",
"AnchorFramesConfig",
"AnchorLatePhaseConfig",
"CeusCaseError",
"CeusConfig",
"ColorPrompt",
"ConstraintsConfig",
"ContrastSide",
"ContrastSideError",
"InteriorDarkRegionConfig",
"KeyframePositionConstraint",
"KeyframesConfig",
"PanSourceConfig",
"PromptFrame",
"PromptObservation",
"PromptSourceConfig",
"ReferenceRegionConfig",
"RoiShapeConfig",
"SmoothingConfig",
"SplitDetectionError",
"TicAnalysisConfig",
"TicBaselineConfig",
"TicFittingConfig",
"VideoFile",
"ViewRect",
"ViewsConfig",
"collect_prompt_observations",
"crop_frame",
"detect_split_x",
"extract_color_prompt",
"local_rect_within",
"locate_contrast_side",
"map_source_mask_to_target",
"mask_iou",
"odd_window",
"optional_positive_int",
"polygon_area",
"polygon_center",
"polygon_to_mask",
"polygons_to_mask",
"prompt_color_mask",
"rect_from_values",
"resample_closed_contour",
"sample_frames_for_split",
"select_prompt_frames",
"split_views",
"target_polygons_to_source",
"triangular_smooth",
"view_offset",
"view_rect",
]
+533
View File
@@ -0,0 +1,533 @@
"""CEUS 定量的参数:`config/ceus.yaml` 里 `ceus` 段的字段定义和校验。
照搬旧项目 `research_config.py` 的配置类,去掉全部默认值:每个字段必填,可以没有的值显式写
null;多写或拼错的键报错;读入后不能修改(`AGENTS.md`「不设默认值」)。各参数的取值与出处写在
`config/ceus.yaml` 的注释里,这里只写字段的含义和校验规则。
旧配置里以下字段删掉了,不留着不用:
- `views.source_view` / `views.target_view`:造影画面在哪一半改由两半各自找绿框决定(`_views.py`);
- `views.effective_region`:裁画面黑边只有旧 `service.py` 用,40 例结果走的 debug 路径不裁;
- `mapping`(含锚点):两半之间的坐标换算改成两个视图矩形左上角之差(`_mapping.py`),
不再需要按某一种视频宽度写死的锚点。
`paths` 段和整份文件最外层的模型由 `scripts/ceus_quant.py` 定义,不在这里。
"""
from typing import Literal, Self
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
_STRICT = ConfigDict(strict=True, extra="forbid", frozen=True)
def _require_positive_odd(value: int, field_name: str) -> int:
"""正奇数校验:时间平滑窗口、自适应阈值块大小都要求奇数(窗口要有中心)。"""
if value < 1:
raise ValueError(f"{field_name} 必须为正整数,收到 {value}")
if value % 2 == 0:
raise ValueError(f"{field_name} 必须为奇数,收到 {value}")
return value
class ViewsConfig(BaseModel):
"""分屏:超声机录造影时屏幕左右各一半,目前只支持左右分屏。
找分屏线只用均匀抽的 `split_sample_frames` 帧,在宽度的 `split_search_band` 比例区间里找。
这两项在旧代码里是写死的数(旧 debug 脚本的 25、旧 `detect_split_x` 参数默认的 (0.35, 0.65)),
这里改成配置,值不变。
"""
model_config = _STRICT
layout: Literal["horizontal_split"]
split_sample_frames: int
split_search_band: list[float]
@field_validator("split_sample_frames")
@classmethod
def _sample_frames_positive(cls, value: int) -> int:
if value < 1:
raise ValueError(f"views.split_sample_frames 必须 ≥1,收到 {value}")
return value
@field_validator("split_search_band")
@classmethod
def _band_pair(cls, value: list[float]) -> list[float]:
if len(value) != 2:
raise ValueError(f"views.split_search_band 必须是 [lo, hi] 两个数,收到 {value}")
return value
class PromptSourceConfig(BaseModel):
"""绿框的 HSV 颜色检测。`min_area` 是候选外接框面积 w×h,不是轮廓面积。"""
model_config = _STRICT
type: Literal["color_polygon"]
hsv_lower: list[int]
hsv_upper: list[int]
ignore_regions: list[list[float]]
min_area: int
class PanSourceConfig(BaseModel):
"""紫圈的 HSV 颜色检测:医生用紫色轮廓圈的一块正常胰腺,作对照。
和绿框是同一类检测器,只是换一组颜色范围。`min_detected_frame_fraction` 是这一组独有的:
紫圈每一帧都在,整段视频里检出的帧占比偏低,说明颜色范围没标定对,对照区整段不可用。
"""
model_config = _STRICT
type: Literal["color_polygon"]
hsv_lower: list[int]
hsv_upper: list[int]
ignore_regions: list[list[float]]
min_area: int
min_detected_frame_fraction: float
@field_validator("hsv_lower", "hsv_upper")
@classmethod
def _hsv_triple(cls, value: list[int]) -> list[int]:
"""HSV 阈值须是三个分量,否则要到 `cv2.inRange` 里才报错,离配置很远。"""
if len(value) != 3:
raise ValueError(f"pan_source HSV 阈值必须是 3 个分量,收到 {value}")
return value
@field_validator("min_area")
@classmethod
def _min_area_positive(cls, value: int) -> int:
"""min_area 须 ≥1。"""
if value < 1:
raise ValueError(f"pan_source.min_area 必须 ≥1,收到 {value}")
return value
@field_validator("min_detected_frame_fraction")
@classmethod
def _fraction_within_bounds(cls, value: float) -> float:
"""占比须在 (0, 1]:取 0 这道门形同虚设,越界则永远判不出有效。"""
if not 0.0 < value <= 1.0:
raise ValueError(f"pan_source.min_detected_frame_fraction 必须 ∈ (0, 1],收到 {value}")
return value
@model_validator(mode="after")
def _hsv_bounds_ordered(self) -> Self:
"""逐分量上界 ≥ 下界。写反了 `cv2.inRange` 不报错,只会恒返回空掩膜。"""
pairs = zip(self.hsv_lower, self.hsv_upper, strict=True)
for index, (low, high) in enumerate(pairs):
if high < low:
raise ValueError(
"pan_source.hsv_upper 每一分量都必须 ≥ hsv_lower,"
f"第 {index} 个分量收到 lower={low}, upper={high}"
)
return self
class KeyframesConfig(BaseModel):
"""提示帧的选择:绿框形状相对上一个提示帧的 IoU 跌破阈值就记一帧新的。
键名沿用旧配置的 `keyframes`;它选的是给分割模型的提示帧,不是第 8 步那 7 张关键帧。
"""
model_config = _STRICT
change_metric: Literal["mask_iou"]
iou_threshold: float
min_gap_frames: int
class SmoothingConfig(BaseModel):
"""病灶轮廓的重采样点数和逐顶点时间平滑窗口(须正奇数)。"""
model_config = _STRICT
contour_points: int
temporal_window: int
@field_validator("temporal_window")
@classmethod
def _temporal_window_odd(cls, value: int) -> int:
"""三角平滑窗口须正奇数,偶数窗口没有中心。"""
return _require_positive_odd(value, "smoothing.temporal_window")
@field_validator("contour_points")
@classmethod
def _contour_points_positive(cls, value: int) -> int:
"""少于 3 个点围不成多边形。"""
if value < 3:
raise ValueError(f"smoothing.contour_points 必须 ≥3,收到 {value}")
return value
class RoiShapeConfig(BaseModel):
"""最终病灶轮廓与绿框的融合(`待定问题/01`)。两个强度截到 [0, 1],面积比须有序。"""
model_config = _STRICT
source: str
keyframe_shape_strength: float
propagation_shape_strength: float
min_area_ratio_to_prompt: float
max_area_ratio_to_prompt: float
@field_validator("keyframe_shape_strength", "propagation_shape_strength")
@classmethod
def _strength_in_unit_interval(cls, value: float) -> float:
"""混合权重截到 [0, 1]。"""
return min(1.0, max(0.0, value))
@model_validator(mode="after")
def _area_ratio_ordered(self) -> Self:
"""面积比下限须小于上限。"""
if self.min_area_ratio_to_prompt >= self.max_area_ratio_to_prompt:
raise ValueError(
"roi_shape.min_area_ratio_to_prompt 必须 < max_area_ratio_to_prompt,"
f"收到 min={self.min_area_ratio_to_prompt}, "
f"max={self.max_area_ratio_to_prompt}"
)
return self
class InteriorDarkRegionConfig(BaseModel):
"""暗区(病灶里不进造影剂、一直发黑的囊液部分)的识别:方法选择、评分和各方法的子参数。"""
model_config = _STRICT
enabled: bool
analysis_view: Literal["source", "target"]
exclude_prompt_color: bool
threshold_method: Literal[
"auto_ensemble",
"otsu",
"percentile",
"absolute",
"adaptive",
"kmeans",
"clahe_otsu",
]
candidate_methods: list[
Literal["otsu", "adaptive", "percentile", "kmeans", "clahe_otsu", "absolute"]
]
threshold_offset: int
min_area: int
min_area_ratio: float
preserve_holes: bool
fill_holes: bool
morphology_kernel: int
fragment_penalty_weight: float
temporal_consistency_weight: float
contour_points: int
percentile: float
threshold: int # absolute 法的固定灰度阈值
adaptive_block_size: int
adaptive_c: float
kmeans_max_iter: int
kmeans_attempts: int
clahe_clip_limit: float
clahe_tile_grid_size: int
@field_validator("adaptive_block_size")
@classmethod
def _adaptive_block_size_odd(cls, value: int) -> int:
"""自适应阈值的块大小须正奇数且 ≥3(OpenCV 的要求)。"""
validated = _require_positive_odd(value, "interior_dark_region.adaptive_block_size")
if validated < 3:
raise ValueError(f"interior_dark_region.adaptive_block_size 必须 ≥3,收到 {validated}")
return validated
@field_validator("percentile")
@classmethod
def _percentile_within_bounds(cls, value: float) -> float:
"""百分位须在 [0, 100]。"""
if not 0.0 <= value <= 100.0:
raise ValueError(f"interior_dark_region.percentile 必须 ∈ [0, 100],收到 {value}")
return value
class KeyframePositionConstraint(BaseModel):
"""提示帧上病灶轮廓的位置约束:把轮廓中心平移向绿框中心。strength 截到 [0, 1]。"""
model_config = _STRICT
enabled: bool
mode: Literal["center_interpolation"]
strength: float
@field_validator("strength")
@classmethod
def _strength_in_unit_interval(cls, value: float) -> float:
"""截到 [0, 1],过冲会把轮廓推出画面。"""
return min(1.0, max(0.0, value))
class ConstraintsConfig(BaseModel):
"""约束段。"""
model_config = _STRICT
keyframe_position: KeyframePositionConstraint
class ReferenceRegionConfig(BaseModel):
"""病灶外背景圆的选取。manual 模式必须给中心坐标,auto_near_roi 时为 null。"""
model_config = _STRICT
enabled: bool
mode: Literal["auto_near_roi", "manual"]
radius_px: int
min_distance_px: int
nonblack_threshold: int
min_valid_fraction: float
center_xy: list[float] | None
@model_validator(mode="after")
def _manual_mode_requires_center(self) -> Self:
"""manual 模式没有中心就取不到背景圆,配置错误直接报错。"""
if self.mode == "manual" and self.center_xy is None:
raise ValueError("reference_region.mode=manual 时 center_xy 必填")
return self
class TicBaselineConfig(BaseModel):
"""增强前基线区间:视频开始到造影剂到达之前那段稳定区间,在上面算基线均值和标准差。"""
model_config = _STRICT
preferred_duration_s: float
minimum_duration_s: float
stability_sd_multiplier: float
@model_validator(mode="after")
def _durations_ordered_and_positive(self) -> Self:
"""两个时长为正且最短 ≤ 首选;写反了「首选取不到退到最短」这条退路就比首选还长。"""
if self.minimum_duration_s <= 0.0 or self.preferred_duration_s <= 0.0:
raise ValueError(
"tic_analysis.baseline 的两个时长必须为正,收到 "
f"preferred={self.preferred_duration_s}, "
f"minimum={self.minimum_duration_s}"
)
if self.minimum_duration_s > self.preferred_duration_s:
raise ValueError(
"tic_analysis.baseline.minimum_duration_s 必须 <= "
f"preferred_duration_s,收到 minimum={self.minimum_duration_s}, "
f"preferred={self.preferred_duration_s}"
)
return self
class TicFittingConfig(BaseModel):
"""TIC 的约束 Lognormal 拟合:候选窗、重复帧门槛、参数边界和质量门。"""
model_config = _STRICT
algorithm_version: Literal["lognormal-v2"]
detect_duplicate_frames: bool
candidate_window_s: float
candidate_long_window_s: float
candidate_polyorder: Literal[2]
max_candidate_gap_s: float
duplicate_mean_absdiff_threshold: float
minimum_valid_samples: int
minimum_peak_distance_s: float
minimum_relative_peak_height: float
multi_peak_minimum_separation_s: float
multi_peak_minimum_relative_height: float
multi_peak_maximum_valley_fraction: float
sigma_min: float
sigma_max: float
robust_loss: Literal["soft_l1"]
max_nfev: int
minimum_r2: float
maximum_nrmse: float
maximum_residual_noise_ratio: float
boundary_margin_s: float
estimated_baseline_percentile: float
estimated_baseline_lookback_s: float
maximum_perf_area_step_fraction: float
parameter_boundary_fraction: float
@field_validator(
"candidate_window_s",
"candidate_long_window_s",
"max_candidate_gap_s",
"duplicate_mean_absdiff_threshold",
"minimum_peak_distance_s",
"multi_peak_minimum_separation_s",
"sigma_min",
"sigma_max",
"max_nfev",
"boundary_margin_s",
"estimated_baseline_lookback_s",
)
@classmethod
def _positive(cls, value: float) -> float:
if value <= 0:
raise ValueError(f"tic_analysis.fitting 的正数参数必须 >0,收到 {value}")
return value
@field_validator("minimum_valid_samples")
@classmethod
def _enough_samples(cls, value: int) -> int:
if value < 8:
raise ValueError("tic_analysis.fitting.minimum_valid_samples 必须 >=8")
return value
@field_validator(
"minimum_relative_peak_height",
"multi_peak_minimum_relative_height",
"multi_peak_maximum_valley_fraction",
"maximum_nrmse",
"maximum_perf_area_step_fraction",
)
@classmethod
def _unit_fraction(cls, value: float) -> float:
if not 0.0 < value <= 1.0:
raise ValueError(f"tic_analysis.fitting 的比例参数必须在 (0,1],收到 {value}")
return value
@field_validator("parameter_boundary_fraction")
@classmethod
def _boundary_fraction(cls, value: float) -> float:
if not 0.0 < value < 0.5:
raise ValueError("tic_analysis.fitting.parameter_boundary_fraction 必须在 (0,0.5) 内")
return value
@field_validator("maximum_residual_noise_ratio")
@classmethod
def _positive_noise_ratio(cls, value: float) -> float:
if value <= 0.0:
raise ValueError("maximum_residual_noise_ratio 必须 >0")
return value
@field_validator("minimum_r2")
@classmethod
def _r2_range(cls, value: float) -> float:
if not -1.0 <= value <= 1.0:
raise ValueError(f"minimum_r2 必须在 [-1,1],收到 {value}")
return value
@field_validator("estimated_baseline_percentile")
@classmethod
def _baseline_percentile(cls, value: float) -> float:
if not 0.0 < value < 50.0:
raise ValueError("estimated_baseline_percentile 必须在 (0,50) 内")
return value
@model_validator(mode="after")
def _sigma_bounds_ordered(self) -> Self:
if self.sigma_min >= self.sigma_max:
raise ValueError("tic_analysis.fitting.sigma_min 必须 < sigma_max")
return self
class AnchorArrivalConfig(BaseModel):
"""造影剂到达判据:门槛取「基线标准差 × 倍数」与「峰高 × 比例」的高者,越线后须持续一段时间。"""
model_config = _STRICT
baseline_sd_multiplier: float
relative_peak_threshold: float
persistence_s: float
@field_validator("baseline_sd_multiplier", "relative_peak_threshold", "persistence_s")
@classmethod
def _positive(cls, value: float) -> float:
"""三个判据数都须为正;0 或负值会让门槛恒真或恒假而不报错。"""
if value <= 0.0:
raise ValueError(f"anchor_frames.arrival 判据数必须为正,收到 {value}")
return value
class AnchorLatePhaseConfig(BaseModel):
"""F2(推注后固定时点的晚期帧)的锚点秒数与容差。"""
model_config = _STRICT
preferred_time_s: float
tolerance_s: float
@field_validator("preferred_time_s", "tolerance_s")
@classmethod
def _positive(cls, value: float) -> float:
"""锚点秒数与容差都须为正。"""
if value <= 0.0:
raise ValueError(f"anchor_frames.late_phase 的数必须为正,收到 {value}")
return value
class AnchorFramesConfig(BaseModel):
"""7 张关键帧的定位参数。
`peak_prominence_baseline_sd_multiplier` 与 `arrival.baseline_sd_multiplier` 是两个独立的数
(即便取值相同),改一个不连带改另一个,所以是两个键。
"""
model_config = _STRICT
arrival: AnchorArrivalConfig
peak_prominence_baseline_sd_multiplier: float
baseline_frame_backoff_s: float
d1_window_after_s: float
d2_window_s: float
d3_window_s: float
d4_window_s: float
b0_window_s: float
late_phase: AnchorLatePhaseConfig
f1_anchor_s: float
@field_validator(
"peak_prominence_baseline_sd_multiplier",
"baseline_frame_backoff_s",
"d1_window_after_s",
"d2_window_s",
"d3_window_s",
"d4_window_s",
"b0_window_s",
"f1_anchor_s",
)
@classmethod
def _positive(cls, value: float) -> float:
"""倍数、退让、容差、锚点秒数都须为正;0 会让窗口退化成一个点。"""
if value <= 0.0:
raise ValueError(f"anchor_frames 的倍数与容差必须为正,收到 {value}")
return value
class TicAnalysisConfig(BaseModel):
"""TIC 四条曲线(病灶整体、病灶去暗区、紫圈、背景圆)的采样、基线、拟合与关键帧。
旧配置里「在曲线最前面插一个时间 0、值 0 的合成点」的三个键已删:那个点不是量出来的,
会在它与首个真实帧之间造出一段最陡的假上升。基线改由 `baseline` 从真实帧里识别。
"""
model_config = _STRICT
brightness_metric: Literal["grayscale_mean"]
exclude_prompt_color: bool
prompt_exclude_dilate: int
baseline: TicBaselineConfig
reference_region: ReferenceRegionConfig
fitting: TicFittingConfig
anchor_frames: AnchorFramesConfig
class CeusConfig(BaseModel):
"""`config/ceus.yaml` 的 `ceus` 段:CEUS 定量的全部算法参数。
`config_version` 标记这一批产物用的是哪一套参数口径,由人改,不从内容哈希派生;
不许是空串。
"""
model_config = _STRICT
config_version: str = Field(min_length=1)
views: ViewsConfig
prompt_source: PromptSourceConfig
pan_source: PanSourceConfig
keyframes: KeyframesConfig
smoothing: SmoothingConfig
roi_shape: RoiShapeConfig
interior_dark_region: InteriorDarkRegionConfig
constraints: ConstraintsConfig
tic_analysis: TicAnalysisConfig
+76
View File
@@ -0,0 +1,76 @@
"""第 2 步:判断造影画面在哪一半,并在那一半上选出提示帧。
哪一半是造影画面,各例不一定相同。绿框只画在造影那一半,所以两半各自按同样的办法
(`collect_prompt_observations` + `select_prompt_frames`,参数相同)选提示帧:恰好一半选出
至少一帧,这一半就是造影画面;两半都选出或都选不出,这一例失败(`ContrastSideError`)。
旧代码按配置写死造影在左半,造影在右半的病例就找不到绿框。
"""
from collections.abc import Sequence
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._prompt_frames import PromptFrame, select_prompt_frames
PROMPT_STEP = 2
type Side = Literal["left", "right"]
class ContrastSideError(CeusCaseError):
"""两半都有或都没有绿框,定不出造影画面在哪一半。"""
def __init__(self, reason: str) -> None:
super().__init__(PROMPT_STEP, reason)
@dataclass(frozen=True)
class ContrastSide:
"""造影画面在哪一半、两个视图矩形(整帧像素坐标),以及在造影那一半上选出的提示帧。
`contrast_view` 是造影那一半(绿框在这里,TIC 在这里采亮度);`target_view` 是另一半
(分割在这里做)。提示帧的掩膜是 `contrast_view` 的局部坐标。
"""
side: Side
contrast_view: ViewRect
target_view: ViewRect
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],
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)
if left and not right:
return ContrastSide("left", left_view, right_view, left)
if right and not left:
return ContrastSide("right", right_view, left_view, right)
if left and right:
raise ContrastSideError(
f"左右两半都找到绿框(左半 {len(left)} 个提示帧,右半 {len(right)} 个),"
"定不出造影画面在哪一半"
)
raise ContrastSideError("左右两半都没有找到绿框,选不出提示帧")
+15
View File
@@ -0,0 +1,15 @@
"""只落在一段视频上的失败。
CEUS 定量第 1–6 步里,某一例的视频不满足算法前提(找不到分屏线、两半都有或都没有绿框等)时
抛 `CeusCaseError`:这一例记进 `failed.csv`,接着跑下一例(`AGENTS.md`「错误怎么处理」)。
其余异常是程序或配置的问题,整次运行停下。
"""
class CeusCaseError(Exception):
"""一例失败:失败在第几步(结构.md 第四部分的 1–9 步)、原因。"""
def __init__(self, step: int, reason: str) -> None:
super().__init__(f"第 {step} 步:{reason}")
self.step = step
self.reason = reason
+202
View File
@@ -0,0 +1,202 @@
"""几何、重采样与平滑的纯函数,供分屏、绿框、提示帧、多边形、TIC 等模块共用。
- `ViewRect`:整数像素坐标的不可变矩形;
- 矩形:`rect_from_values`、`crop_frame`、`local_rect_within`;
- 掩膜:`mask_iou`、`polygon_to_mask`、`polygons_to_mask`;
- 多边形:`polygon_area`、`polygon_center`、`resample_closed_contour`;
- 时间平滑:`triangular_smooth`;配置解析:`odd_window`、`optional_positive_int`。
算法等价于伙伴参考组件 `references/ROI/roi_tracking_pipeline.py`(经旧项目重写)。
"""
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any
import cv2
import numpy as np
import numpy.typing as npt
type Mask = npt.NDArray[np.bool_]
type Points = npt.NDArray[np.float32]
type Frame = npt.NDArray[np.uint8]
type AnyArray = npt.NDArray[Any]
@dataclass(frozen=True)
class ViewRect:
"""整数像素坐标的矩形,左闭右开 `[x1, x2) × [y1, y2)`。
构造时强制 `0 <= x1 < x2` 且 `0 <= y1 < y2`,非法矩形直接报错,不让退化或负坐标往下游走。
"""
x1: int
y1: int
x2: int
y2: int
def __post_init__(self) -> None:
if not (0 <= self.x1 < self.x2 and 0 <= self.y1 < self.y2):
raise ValueError(
f"非法 ViewRect:要求 0<=x1<x2 且 0<=y1<y2,得到 "
f"({self.x1}, {self.y1}, {self.x2}, {self.y2})。"
)
@property
def width(self) -> int:
"""宽度 `x2 - x1`。"""
return self.x2 - self.x1
@property
def height(self) -> int:
"""高度 `y2 - y1`。"""
return self.y2 - self.y1
def mask_iou(mask_a: AnyArray, mask_b: AnyArray) -> float:
"""两个二值掩膜的交并比;并集为空时返回 0.0。"""
intersection = np.logical_and(mask_a, mask_b).sum()
union = np.logical_or(mask_a, mask_b).sum()
return float(intersection / union) if union else 0.0
def crop_frame[T: np.generic](frame: npt.NDArray[T], rect: ViewRect) -> npt.NDArray[T]:
"""按矩形切出子图 `frame[y1:y2, x1:x2]`(视图切片,不复制)。"""
return frame[rect.y1 : rect.y2, rect.x1 : rect.x2]
def local_rect_within(parent: ViewRect, child: ViewRect) -> ViewRect:
"""把 `child` 换算成相对 `parent` 左上角的坐标;越界时 `ViewRect` 构造报错。"""
return ViewRect(
child.x1 - parent.x1,
child.y1 - parent.y1,
child.x2 - parent.x1,
child.y2 - parent.y1,
)
def rect_from_values(values: Sequence[float], width: int, height: int) -> ViewRect:
"""把 `[x1, y1, x2, y2]` 解析成像素矩形。
四个值都在 [0, 1] 时按归一化坐标乘宽高,否则按像素取整。结果裁到画面内,并保证
`x2 >= x1 + 1`、`y2 >= y1 + 1`,不出退化矩形。
"""
if len(values) != 4:
raise ValueError("手动裁剪坐标必须是 [x1, y1, x2, y2] 四元组。")
if all(0.0 <= float(value) <= 1.0 for value in values):
scales = (width, height, width, height)
x1, y1, x2, y2 = (
round(float(value) * scale) for value, scale in zip(values, scales, strict=True)
)
else:
x1, y1, x2, y2 = (round(float(value)) for value in values)
x1 = int(np.clip(x1, 0, width - 1))
y1 = int(np.clip(y1, 0, height - 1))
x2 = int(np.clip(x2, x1 + 1, width))
y2 = int(np.clip(y2, y1 + 1, height))
return ViewRect(x1, y1, x2, y2)
def polygon_to_mask(polygon: AnyArray, height: int, width: int) -> Mask:
"""把多边形(`(N, 2)`,x、y 像素坐标)栅格化为 `(height, width)` 的布尔掩膜。"""
canvas = np.zeros((height, width), dtype=np.uint8)
points = np.round(polygon).astype(np.int32).reshape(-1, 1, 2)
cv2.fillPoly(canvas, [points], (255,))
return canvas > 0
def polygons_to_mask(polygons: Sequence[AnyArray], height: int, width: int) -> Mask:
"""把「外轮廓 + 洞」的多边形列表一次栅格化成带孔掩膜。
整个列表放进同一次 `cv2.fillPoly`:一次调用里的多条轮廓按奇偶规则填充,被包住的内轮廓
成为孔。逐个多边形各自栅格化再取并集会把洞当成外轮廓填回去,暗区面积就虚大了。
顶点数少于 3 的项跳过;全部退化时返回全 False。
"""
canvas = np.zeros((height, width), dtype=np.uint8)
contours = [
np.round(polygon).astype(np.int32).reshape(-1, 1, 2)
for polygon in polygons
if len(polygon) >= 3
]
if contours:
cv2.fillPoly(canvas, contours, (255,))
return canvas > 0
def polygon_area(polygon: AnyArray) -> float:
"""鞋带公式算多边形面积(像素²)。直接在浮点顶点上算、不取整,避免量化误差。"""
points = np.asarray(polygon, dtype=np.float64)
x = points[:, 0]
y = points[:, 1]
cross = x * np.roll(y, -1) - np.roll(x, -1) * y
return abs(0.5 * float(cross.sum()))
def polygon_center(polygon: AnyArray) -> Points:
"""多边形质心(一阶矩);面积退化到约 0 时取顶点均值,避免除零。"""
contour = np.round(polygon).astype(np.float32).reshape(-1, 1, 2)
moments = cv2.moments(contour)
if abs(moments["m00"]) > 1e-6:
return np.array(
[moments["m10"] / moments["m00"], moments["m01"] / moments["m00"]],
dtype=np.float32,
)
return np.asarray(polygon.mean(axis=0), dtype=np.float32)
def resample_closed_contour(contour: AnyArray | None, num_points: int) -> Points | None:
"""沿弧长把闭合轮廓等距重采样到 `num_points` 个点。
轮廓为 None、顶点少于 3 个或周长不超过 1 像素时返回 None。
"""
if contour is None or len(contour) < 3:
return None
points = np.asarray(contour, dtype=np.float32)
closed = np.vstack([points, points[0]])
deltas = np.diff(closed, axis=0)
segment_lengths = np.sqrt((deltas * deltas).sum(axis=1))
perimeter = float(segment_lengths.sum())
if perimeter <= 1.0:
return None
cumulative = np.concatenate([[0.0], np.cumsum(segment_lengths)])
samples = np.linspace(0.0, perimeter, num_points, endpoint=False)
resampled = np.zeros((num_points, 2), dtype=np.float32)
resampled[:, 0] = np.interp(samples, cumulative, closed[:, 0])
resampled[:, 1] = np.interp(samples, cumulative, closed[:, 1])
return resampled
def triangular_smooth(values: AnyArray, window: int) -> AnyArray:
"""沿第 0 轴(时间)做三角加权滑动平均。
权重 `1, 2, …, r+1, …, 2, 1` 归一化,边界复制端值。窗口 ≤1 时原样返回副本;偶数窗口没有
中心,报错。
"""
if window <= 1:
return values.copy()
if window % 2 == 0:
raise ValueError("时间平滑窗口必须为奇数。")
radius = window // 2
ramp = np.arange(1, radius + 2, dtype=np.float32)
weights = np.concatenate([ramp, ramp[-2::-1]])
weights /= weights.sum()
flat = values.reshape(values.shape[0], -1)
padded = np.pad(flat, ((radius, radius), (0, 0)), mode="edge")
smoothed = np.zeros_like(flat)
for index in range(flat.shape[0]):
smoothed[index] = np.sum(padded[index : index + window] * weights[:, None], axis=0)
return smoothed.reshape(values.shape)
def odd_window(value: int) -> int:
"""把整数变成不小于 1 的奇数:偶数加 1,小于 1 的取 1。"""
floored = max(int(value), 1)
return floored if floored % 2 == 1 else floored + 1
def optional_positive_int(value: Any) -> int | None:
"""可选的正整数配置:None、"auto" 或不大于 0 的值都返回 None,其余按整数解析。"""
if value is None or value == "auto":
return None
parsed = int(value)
return parsed if parsed > 0 else None
+186
View File
@@ -0,0 +1,186 @@
"""第 2 步的绿框检测:在已切出的一半画面上找医生画的彩色轮廓。
- `prompt_color_mask`:TIC 采样时要从亮度统计里剔除医生画上去的线,这里给出那组颜色的掩膜,
受 `exclude_prompt_color` 开关和 `prompt_exclude_dilate` 膨胀核控制;
- `extract_color_prompt`:HSV 范围内的像素 → 抹掉顶部文字带 → 形态学闭运算加膨胀把断开的描边
连起来 → 外轮廓 → 按外接框面积 w×h 过滤、取最大者 → 实心填充成掩膜;没有候选返回 None;
- `collect_prompt_observations`:逐帧调上一个函数,没检出的帧用 None 占位,帧号与下标对齐。
颜色来自参数:喂绿框那组配置检绿框,喂紫圈那组配置(`PanSourceConfig`)检紫圈。帧是已经切出的
一半画面,返回的外接框和掩膜都是这一半的局部坐标。
"""
from collections.abc import Iterable, 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._geometry import AnyArray, Frame, Mask, rect_from_values
# 形态学结构元:5×5,闭运算与膨胀各一次。
_MORPH_KERNEL = np.ones((5, 5), dtype=np.uint8)
class _ColorHsvConfigLike(Protocol):
"""一组 HSV 颜色范围(`PromptSourceConfig`、`PanSourceConfig` 都符合)。"""
@property
def hsv_lower(self) -> list[int]: ...
@property
def hsv_upper(self) -> list[int]: ...
class _ColorContourConfigLike(_ColorHsvConfigLike, Protocol):
"""一整套彩色轮廓检测配置:HSV 范围、忽略区、面积下限。"""
@property
def ignore_regions(self) -> list[list[float]]: ...
@property
def min_area(self) -> int: ...
@dataclass(frozen=True)
class ColorPrompt:
"""一帧的检测结果(这一半画面的局部坐标)。
`mask` 是最大候选轮廓实心填充的布尔掩膜;`contour` 是 `cv2.findContours` 原样的 `(N, 1, 2)`
int32 轮廓点;`bbox_xyxy` 是含端点的外接框(`x2 = x + w - 1`);`bbox_area` 是外接框面积 w×h,
用于 `min_area` 过滤和排序;`contour_area` 只在外接框面积相同时作次级排序;`candidate_count`
是通过 `min_area` 的候选数。
"""
mask: Mask
contour: AnyArray
bbox_xyxy: tuple[int, int, int, int]
bbox_area: int
contour_area: float
candidate_count: int
@property
def width(self) -> int:
"""含端点的外接框宽度 `x2 - x1 + 1`。"""
return self.bbox_xyxy[2] - self.bbox_xyxy[0] + 1
@property
def height(self) -> int:
"""含端点的外接框高度 `y2 - y1 + 1`。"""
return self.bbox_xyxy[3] - self.bbox_xyxy[1] + 1
@dataclass(frozen=True)
class PromptObservation:
"""一帧的观测:帧号和检测结果(没检出时为 None)。"""
frame_index: int
prompt: ColorPrompt | None
@property
def detected(self) -> bool:
"""这一帧是否检出。"""
return self.prompt is not None
def prompt_color_mask(
source_bgr: Frame,
color_cfg: _ColorHsvConfigLike,
tic_cfg: TicAnalysisConfig,
) -> Mask:
"""TIC 采样时要剔除的那组颜色的像素掩膜。
`exclude_prompt_color` 关闭时返回全 False(不剔除任何像素);开启时取 HSV 范围内的像素,
`prompt_exclude_dilate` 大于 1 时再按这个核大小膨胀一次(核为 1 时膨胀是恒等操作)。
"""
if not tic_cfg.exclude_prompt_color:
return np.zeros(source_bgr.shape[:2], dtype=bool)
lower = np.asarray(color_cfg.hsv_lower, dtype=np.uint8)
upper = np.asarray(color_cfg.hsv_upper, dtype=np.uint8)
hsv = cv2.cvtColor(source_bgr, cv2.COLOR_BGR2HSV)
mask: Mask = np.asarray(cv2.inRange(hsv, lower, upper)) > 0
kernel_size = tic_cfg.prompt_exclude_dilate
if kernel_size > 1:
kernel = np.ones((kernel_size, kernel_size), dtype=np.uint8)
mask = np.asarray(cv2.dilate(mask.astype(np.uint8) * 255, kernel, iterations=1)) > 0
return mask
def _apply_ignore_regions(mask: AnyArray, regions: Sequence[Sequence[float]]) -> None:
"""把忽略区(归一化或像素坐标的矩形)在掩膜上原地置 0,用来抹掉顶部文字带。"""
height, width = mask.shape
for region in regions:
rect = rect_from_values(list(region), width, height)
mask[rect.y1 : rect.y2, rect.x1 : rect.x2] = 0
@dataclass(frozen=True)
class _Candidate:
contour: AnyArray
bbox_xyxy: tuple[int, int, int, int]
bbox_area: int
contour_area: float
def _largest_bbox_candidate(
contours: Sequence[AnyArray], min_area: int
) -> tuple[_Candidate, int] | None:
"""按外接框面积过滤、取最大的候选;返回它和候选总数,没有合格候选时返回 None。"""
candidates: list[_Candidate] = []
for contour in contours:
x, y, w, h = cv2.boundingRect(contour)
bbox_area = int(w * h)
if bbox_area < min_area:
continue
candidates.append(
_Candidate(
contour=contour,
bbox_xyxy=(int(x), int(y), int(x + w - 1), int(y + h - 1)),
bbox_area=bbox_area,
contour_area=float(cv2.contourArea(contour)),
)
)
if not candidates:
return None
candidates.sort(key=lambda item: (item.bbox_area, item.contour_area), reverse=True)
return candidates[0], len(candidates)
def extract_color_prompt(
source_bgr: Frame, prompt_cfg: _ColorContourConfigLike
) -> ColorPrompt | None:
"""在一半画面的一帧上找医生画的彩色轮廓,返回最大的合格候选;没有返回 None。"""
lower = np.asarray(prompt_cfg.hsv_lower, dtype=np.uint8)
upper = np.asarray(prompt_cfg.hsv_upper, dtype=np.uint8)
hsv = cv2.cvtColor(source_bgr, cv2.COLOR_BGR2HSV)
mask = np.asarray(cv2.inRange(hsv, lower, upper), dtype=np.uint8)
_apply_ignore_regions(mask, prompt_cfg.ignore_regions)
mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, _MORPH_KERNEL, iterations=1)
mask = cv2.dilate(mask, _MORPH_KERNEL, iterations=1)
contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
selected = _largest_bbox_candidate([np.asarray(c) for c in contours], prompt_cfg.min_area)
if selected is None:
return None
best, candidate_count = selected
filled = np.zeros(mask.shape, dtype=np.uint8)
cv2.drawContours(filled, [best.contour], -1, (255,), thickness=-1)
return ColorPrompt(
mask=filled > 0,
contour=best.contour,
bbox_xyxy=best.bbox_xyxy,
bbox_area=best.bbox_area,
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)
]
+67
View File
@@ -0,0 +1,67 @@
"""两半画面之间的坐标换算。
绿框画在造影那一半;分割在另一半(灰阶画面,病灶边界不随造影剂进出明暗变化)上做,轮廓再挪回
造影那一半采亮度。两半是同一个探头视野的两路成像,大小相同、只差一个水平平移,所以换算只做平移:
平移量 = 目标那一半视图的左上角 − 造影那一半视图的左上角(整帧像素坐标)
两个视图矩形直接来自第 1 步的分屏结果,不依赖视频宽度。旧代码改用按 1856 像素宽的视频写死的
锚点,其余宽度的视频会挪偏一两百像素(design 0002 决定 4),这里不再用锚点。
两个函数都在各自视图的局部坐标里收发掩膜和多边形:局部 → 整帧(加本视图左上角)→ 平移 →
目标视图局部(减目标视图左上角)。两半宽度若差几个像素,超出目标视图的部分裁掉,不缩放。
"""
import numpy as np
from cadx.ceus._geometry import AnyArray, Mask, Points, ViewRect
def view_offset(source_view_rect: ViewRect, target_view_rect: ViewRect) -> tuple[int, int]:
"""从造影那一半到目标那一半的平移量 `(dx, dy)`:两个视图左上角之差。"""
return (
target_view_rect.x1 - source_view_rect.x1,
target_view_rect.y1 - source_view_rect.y1,
)
def map_source_mask_to_target(
source_mask: AnyArray,
source_view_rect: ViewRect,
target_view_rect: ViewRect,
) -> Mask:
"""把造影那一半局部坐标的掩膜挪到目标那一半的局部坐标,形状为目标视图大小。"""
dx, dy = view_offset(source_view_rect, target_view_rect)
# 源局部 (x, y) → 整帧 (x + sx1, y + sy1) → 平移 → 目标局部 (x + sx1 + dx − tx1, …)。
shift_x = source_view_rect.x1 + dx - target_view_rect.x1
shift_y = source_view_rect.y1 + dy - target_view_rect.y1
mapped = np.zeros((target_view_rect.height, target_view_rect.width), dtype=bool)
ys, xs = np.nonzero(source_mask)
xs = xs + shift_x
ys = ys + shift_y
inside = (xs >= 0) & (xs < target_view_rect.width) & (ys >= 0) & (ys < target_view_rect.height)
mapped[ys[inside], xs[inside]] = True
return mapped
def target_polygons_to_source(
target_polygons: list[AnyArray],
source_view_rect: ViewRect,
target_view_rect: ViewRect,
) -> list[Points]:
"""把目标那一半局部坐标的多边形逐个挪回造影那一半的局部坐标,裁到视图边界内。"""
dx, dy = view_offset(source_view_rect, target_view_rect)
shift = np.array(
[
target_view_rect.x1 - dx - source_view_rect.x1,
target_view_rect.y1 - dy - source_view_rect.y1,
],
dtype=np.float32,
)
mapped: list[Points] = []
for polygon in target_polygons:
source_poly = polygon.astype(np.float32) + shift[None, :]
source_poly[:, 0] = np.clip(source_poly[:, 0], 0, source_view_rect.width - 1)
source_poly[:, 1] = np.clip(source_poly[:, 1], 0, source_view_rect.height - 1)
mapped.append(source_poly)
return mapped
+78
View File
@@ -0,0 +1,78 @@
"""第 2 步的提示帧选择:绿框位置或形状明显变化的那几帧,交给分割模型当「病灶在这里」的提示。
这批帧不是第 8 步的 7 张关键帧。规则:
- 第一个检出绿框的帧必选,`change_iou` 记 1.0;
- 之后每帧先过最小间隔门(距上一个提示帧不足 `min_gap_frames` 帧的跳过),再算它的绿框掩膜与
上一个提示帧的 IoU,严格小于 `iou_threshold` 才记新提示帧(恰好等于不记);
- 没检出绿框的帧跳过,不占第一帧的位置。
"""
from dataclasses import dataclass
from cadx.ceus._config import KeyframesConfig
from cadx.ceus._geometry import Mask, mask_iou
from cadx.ceus._green_prompt import PromptObservation
# 间隔门的初值:保证第一个检出绿框的帧总能通过。
_NO_PREVIOUS_PROMPT_FRAME = -(10**9)
@dataclass(frozen=True)
class PromptFrame:
"""一个提示帧(这一半画面的局部坐标)。
`mask` 是这一帧绿框的实心掩膜,既是给分割的提示,也是下一个提示帧比 IoU 的基准;
`change_iou` 是相对上一个提示帧的 IoU,第一帧记 1.0。
"""
frame_index: int
mask: Mask
bbox_xyxy: tuple[int, int, int, int]
bbox_area: int
contour_area: float
change_iou: float
def select_prompt_frames(
observations: list[PromptObservation], prompt_frame_cfg: KeyframesConfig
) -> list[PromptFrame]:
"""从逐帧观测中选出提示帧,按帧号升序;没有任何检出时返回空列表。"""
if prompt_frame_cfg.change_metric != "mask_iou":
raise RuntimeError(
f"keyframes.change_metric 仅支持 'mask_iou',得到 '{prompt_frame_cfg.change_metric}'。"
)
iou_threshold = prompt_frame_cfg.iou_threshold
min_gap = prompt_frame_cfg.min_gap_frames
prompt_frames: list[PromptFrame] = []
last_key_mask: Mask | None = None
last_key_frame = _NO_PREVIOUS_PROMPT_FRAME
for observation in observations:
prompt = observation.prompt
if prompt is None:
continue
frame_index = observation.frame_index
if frame_index - last_key_frame < min_gap:
continue
if last_key_mask is None:
change_iou = 1.0
is_new_prompt_frame = True
else:
change_iou = mask_iou(last_key_mask, prompt.mask)
is_new_prompt_frame = change_iou < iou_threshold
if not is_new_prompt_frame:
continue
prompt_frames.append(
PromptFrame(
frame_index=frame_index,
mask=prompt.mask,
bbox_xyxy=prompt.bbox_xyxy,
bbox_area=prompt.bbox_area,
contour_area=prompt.contour_area,
change_iou=change_iou,
)
)
last_key_mask = prompt.mask
last_key_frame = frame_index
return prompt_frames
+186
View File
@@ -0,0 +1,186 @@
"""第 1 步的分屏:找出左右两半之间的竖直暗缝,切出两个视图矩形。
超声机录造影时屏幕左右各一半,分界不一定正好在正中间,所以从帧内容里找:在中部一段列范围内
取列均值最低、且明显低于周围的那一列。找不到可信的缝就报错,不退回 `W//2`。
照旧 debug 路径(40 例结果是它跑出来的):找缝只用均匀抽的若干帧;不裁画面黑边,视图矩形就是
左右两半本身。
"""
from collections.abc import Iterable, Mapping, Sequence
import cv2
import numpy as np
import numpy.typing as npt
from cadx.ceus._config import ViewsConfig
from cadx.ceus._errors import CeusCaseError
from cadx.ceus._geometry import Frame, ViewRect, rect_from_values
SPLIT_STEP = 1
class SplitDetectionError(CeusCaseError):
"""这段视频找不到可信的分屏线(无帧、无暗缝、切分越界或左右失衡)。"""
def __init__(self, reason: str) -> None:
super().__init__(SPLIT_STEP, reason)
def _mean_gray(frames: Iterable[Frame]) -> npt.NDArray[np.float32]:
"""若干帧取平均后转灰度。边收边累加、不把帧堆成一个数组;累加器用 float64,帧多时不丢低位。"""
total: npt.NDArray[np.float64] | None = None
count = 0
for frame in frames:
if total is None:
total = np.zeros(frame.shape, dtype=np.float64)
elif frame.shape != total.shape:
raise SplitDetectionError(
f"帧尺寸不一致:首帧 {total.shape},第 {count} 帧 {frame.shape}。"
)
total += frame
count += 1
if total is None:
raise SplitDetectionError("帧序列为空,无法求跨帧平均灰度。")
mean_bgr = (total / count).astype(np.uint8)
return np.asarray(cv2.cvtColor(mean_bgr, cv2.COLOR_BGR2GRAY), dtype=np.float32)
def detect_split_x(frames: Iterable[Frame], *, search_band: tuple[float, float]) -> int:
"""找左右分屏的竖直分界列 `split_x`。
`search_band` 是搜索区间占宽度的比例 `(lo, hi)`,只在中部找,排除画面内偶发的暗带。
在区间内取列均值最低的列;它的列均值须低于区间中位列均值的 70%(否则视为没有缝,
例如整幅平坦或全黑);再要求左右宽度比在 [0.5, 2.0]。任一不满足报 `SplitDetectionError`。
"""
lo_ratio, hi_ratio = search_band
if not 0.0 <= lo_ratio < hi_ratio <= 1.0:
raise SplitDetectionError(f"非法 search_band:要求 0<=lo<hi<=1,收到 {search_band}。")
column_means = _mean_gray(frames).mean(axis=0)
width = int(column_means.shape[0])
band_lo = round(lo_ratio * width)
band_hi = round(hi_ratio * width)
if band_hi - band_lo < 1:
raise SplitDetectionError(
f"search_band 在宽度 {width} 下退化为空区间 [{band_lo}, {band_hi})。"
)
band_means = column_means[band_lo:band_hi]
split_x = band_lo + int(np.argmin(band_means))
band_median = float(np.median(band_means))
valley_value = float(column_means[split_x])
if valley_value >= band_median * 0.7:
raise SplitDetectionError(
f"中部 band 内未找到稳定暗缝(谷值 {valley_value:.1f} 未显著低于 "
f"中位 {band_median:.1f}),疑为无缝或全黑画面。"
)
if not band_lo <= split_x < band_hi:
raise SplitDetectionError(
f"检测到的 split_x={split_x} 落在搜索 band [{band_lo}, {band_hi}) 外。"
)
left_width = split_x
right_width = width - split_x
if left_width < 1 or right_width < 1:
raise SplitDetectionError(
f"切分线 split_x={split_x} 使某一侧宽度退化(左 {left_width} / 右 {right_width})。"
)
width_ratio = left_width / right_width
if not 0.5 <= width_ratio <= 2.0:
raise SplitDetectionError(
f"左右宽度比 {width_ratio:.2f} 超出 [0.5, 2.0]"
f"(左 {left_width} / 右 {right_width}),切分线不可信。"
)
return split_x
def _horizontal_split_rect(
view_name: str, width: int, height: int, split_x: int | None
) -> ViewRect | None:
"""左右分屏的视图矩形,分界用检测到的 `split_x`。"""
if split_x is None:
raise ValueError("horizontal_split 布局必须传入检测到的 split_x。")
if view_name == "left":
return ViewRect(0, 0, split_x, height)
if view_name == "right":
return ViewRect(split_x, 0, width, height)
return None
def _vertical_split_rect(view_name: str, width: int, height: int) -> ViewRect | None:
"""上下分屏的视图矩形(分界取 H//2)。"""
mid = height // 2
if view_name == "top":
return ViewRect(0, 0, width, mid)
if view_name == "bottom":
return ViewRect(0, mid, width, height)
return None
def _manual_crop_rect(
view_cfg: Mapping[str, object], view_name: str, width: int, height: int
) -> ViewRect:
"""手工裁剪布局:从 `crops` 取对应视图的框,没有就报错。"""
crops = view_cfg["crops"]
if not isinstance(crops, Mapping) or view_name not in crops:
raise RuntimeError(f"手动裁剪中未找到视图 {view_name!r}。")
values: object = crops[view_name] # pyright: ignore[reportUnknownVariableType]
if not isinstance(values, list):
raise RuntimeError(f"手动裁剪视图 {view_name!r} 必须是 [x1, y1, x2, y2]。")
return rect_from_values([float(v) for v in values], width, height) # pyright: ignore[reportUnknownVariableType, reportUnknownArgumentType]
def view_rect(
view_cfg: Mapping[str, object],
view_name: str,
width: int,
height: int,
*,
split_x: int | None,
) -> ViewRect:
"""按布局和视图名算视图矩形(整帧像素坐标)。
`view_cfg` 至少含 `layout`,手工裁剪时含 `crops`。左右分屏时 `split_x` 必填。
布局与视图名对不上报 `RuntimeError`。
"""
layout = view_cfg["layout"]
rect: ViewRect | None
if layout == "horizontal_split":
rect = _horizontal_split_rect(view_name, width, height, split_x)
elif layout == "vertical_split":
rect = _vertical_split_rect(view_name, width, height)
elif layout == "full":
rect = ViewRect(0, 0, width, height)
elif layout == "manual_crop":
rect = _manual_crop_rect(view_cfg, view_name, width, height)
else:
rect = None
if rect is None:
raise RuntimeError(f"不支持的 layout/view 组合:{layout}/{view_name}")
return rect
def sample_frames_for_split(frames: Sequence[Frame], sample_count: int) -> list[Frame]:
"""沿时间轴均匀抽 `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]
def split_views(frames: Sequence[Frame], views_cfg: ViewsConfig) -> tuple[int, ViewRect, ViewRect]:
"""第 1 步:在均匀抽的帧上找分屏线,交出 `(split_x, 左半矩形, 右半矩形)`。"""
split_x = detect_split_x(
sample_frames_for_split(frames, views_cfg.split_sample_frames),
search_band=(views_cfg.split_search_band[0], views_cfg.split_search_band[1]),
)
height, width = frames[0].shape[:2]
layout = {"layout": views_cfg.layout}
left = view_rect(layout, "left", width, height, split_x=split_x)
right = view_rect(layout, "right", width, height, split_x=split_x)
return split_x, left, right
+90
View File
@@ -0,0 +1,90 @@
"""按路径用 cv2 逐帧读视频。
反复遍历靠重新解码,不缓存帧:一段 1260 帧的素材解一遍不到两秒,缓存下来要好几 GB(旧项目实测)。
每次遍历各开一个 `cv2.VideoCapture`,互不干扰。探测不到有效 fps、帧数或尺寸直接报错,不回退默认值。
"""
import math
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
import cv2
import numpy as np
import numpy.typing as npt
from cadx.ceus._geometry import Frame
@dataclass(frozen=True)
class VideoFile:
"""磁盘上的一段视频,可以反复从第 0 帧遍历。"""
path: Path
def _capture(self) -> cv2.VideoCapture:
cap = cv2.VideoCapture(str(self.path))
if not cap.isOpened():
cap.release()
raise RuntimeError(f"cv2 打不开视频:{self.path}")
return cap
def probe(self) -> tuple[float, int, int, int]:
"""视频元数据 `(fps, frame_count, width, height)`,都为正,否则报错。"""
cap = self._capture()
try:
fps = float(cap.get(cv2.CAP_PROP_FPS))
frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
finally:
cap.release()
if not math.isfinite(fps) or fps <= 0:
raise RuntimeError(f"探测到非法 fps={fps}:{self.path}")
if frame_count <= 0 or width <= 0 or height <= 0:
raise RuntimeError(
f"探测到非法元数据 frame_count={frame_count} "
f"width={width} height={height}:{self.path}"
)
return fps, frame_count, width, height
def iter_frames(self) -> Iterator[tuple[int, Frame]]:
"""逐帧产出 `(帧号, BGR 的 H×W×3 uint8)`,帧号从 0 连续递增;每次调用都从头开始。"""
cap = self._capture()
try:
index = 0
while True:
ok, frame = cap.read()
if not ok:
break
yield index, np.asarray(frame, dtype=np.uint8)
index += 1
finally:
cap.release()
def frame_times_s(self) -> npt.NDArray[np.float64]:
"""逐帧读容器里的显示时间戳(秒);必须严格递增、帧数与容器记录一致,否则报错。"""
cap = self._capture()
try:
expected_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
times: list[float] = []
while True:
ok, _frame = cap.read()
if not ok:
break
timestamp_s = float(cap.get(cv2.CAP_PROP_POS_MSEC)) / 1000.0
if not math.isfinite(timestamp_s) or timestamp_s < 0.0:
raise RuntimeError(f"视频包含非法显示时间戳 {timestamp_s}:{self.path}")
if times and timestamp_s <= times[-1]:
raise RuntimeError(
"视频显示时间戳不严格递增:"
f"第 {len(times)} 帧 {timestamp_s},前一帧 {times[-1]}:{self.path}"
)
times.append(timestamp_s)
finally:
cap.release()
if len(times) != expected_count:
raise RuntimeError(
f"时间戳帧数与容器记录的帧数不一致:{len(times)} / {expected_count}:{self.path}"
)
return np.asarray(times, dtype=np.float64)
+2 -1
View File
@@ -13,7 +13,7 @@ from cadx.library._read import (
read_ultrasound,
)
from cadx.library._scan import LeakHit, scan_for_leaks
from cadx.library._write import LibraryCase, write_library
from cadx.library._write import LibraryCase, check_key_image, write_library
__all__ = [
"ClinicalRoute",
@@ -21,6 +21,7 @@ __all__ = [
"LibraryCase",
"MrRoute",
"UltrasoundRoute",
"check_key_image",
"list_aliases",
"read_clinical",
"read_mr",
+22 -12
View File
@@ -56,23 +56,33 @@ def _write_json(path: Path, content: object) -> None:
path.write_text(json.dumps(content, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def check_key_image(path: Path) -> str | None:
"""关键图能逐像素保真地重编码时返回 None,不能时返回原因。
不能保真的:不是 PNG、位深或颜色类型不在 `_FAITHFUL` 里、带 tRNS 透明色。
build 在写库前对每例调用,返回原因的整例跳过(C14),不悄悄降位或丢透明。
"""
header = path.read_bytes()[:26]
if header[:8] != _PNG_SIGNATURE or header[12:16] != b"IHDR":
return "不是 PNG"
bit_depth, color_type = header[24], header[25]
if (bit_depth, color_type) not in _FAITHFUL:
return f"位深 {bit_depth}、颜色类型 {color_type} 不能逐像素保真重编码"
with Image.open(path) as image:
if "transparency" in image.info:
return "带 tRNS 透明色,重编码会丢透明"
return None
def _write_clean_png(source: Path, target: Path) -> None:
"""只保留像素和颜色模式重新编码:文本块、EXIF、ICC 等附加数据都不带过去。
不能逐像素保真的图(非 PNG、位深或颜色类型不在 `_FAITHFUL` 里、带 tRNS 透明色)报错停下,
不悄悄降位或丢透明。
走到这里的图应当已由 build 用 `check_key_image` 查过;还不能保真说明 build 漏了检查,报错停下。
"""
header = source.read_bytes()[:26]
if header[:8] != _PNG_SIGNATURE or header[12:16] != b"IHDR":
raise ValueError(f"关键图不是 PNG:{source}")
bit_depth, color_type = header[24], header[25]
if (bit_depth, color_type) not in _FAITHFUL:
raise ValueError(
f"关键图位深 {bit_depth}、颜色类型 {color_type} 不能逐像素保真重编码:{source}"
)
reason = check_key_image(source)
if reason is not None:
raise ValueError(f"关键图{reason},build 应已用 check_key_image 跳过这一例:{source}")
with Image.open(source) as image:
if "transparency" in image.info:
raise ValueError(f"关键图带 tRNS 透明色,重编码会丢透明:{source}")
clean = Image.frombytes(image.mode, image.size, image.tobytes())
clean.save(target, format="PNG")