iomgaa
1de88a5d0a
- 内存：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>
46 lines
1.9 KiB
Python
46 lines
1.9 KiB
Python
"""runs 的快照收尾：`finish_snapshot` 只新建 finished.json，不改开跑时写的快照，也不收两次尾。"""
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from cadx.runs import GitState, finish_snapshot, write_snapshot
|
|
|
|
_GIT = GitState(commit="a" * 40, short_commit="aaaaaaa", clean=True, diff_sha256="b" * 64)
|
|
|
|
|
|
def _snapshot(tmp_path: Path) -> Path:
|
|
lock = tmp_path / "uv.lock"
|
|
lock.write_text("lock", encoding="utf-8")
|
|
config = tmp_path / "x.yaml"
|
|
config.write_text("a: 1\n", encoding="utf-8")
|
|
snapshot_dir = tmp_path / "snapshot"
|
|
started = datetime(2026, 10, 16, 9, 0, 0).astimezone()
|
|
write_snapshot(snapshot_dir, "run-1", _GIT, started, lock, {"x.yaml": config})
|
|
return snapshot_dir
|
|
|
|
|
|
def test_finish_writes_separate_file_and_keeps_snapshot(tmp_path: Path) -> None:
|
|
snapshot_dir = _snapshot(tmp_path)
|
|
before = (snapshot_dir / "snapshot.json").read_bytes()
|
|
finished = datetime(2026, 10, 16, 10, 0, 0).astimezone()
|
|
finish_snapshot(snapshot_dir, finished, {"segment_cache_hits": 3})
|
|
assert (snapshot_dir / "snapshot.json").read_bytes() == before
|
|
record = json.loads((snapshot_dir / "finished.json").read_text(encoding="utf-8"))
|
|
assert record == {"finished_at": finished.isoformat(), "segment_cache_hits": 3}
|
|
|
|
|
|
def test_finish_twice_or_with_finished_at_in_record_raises(tmp_path: Path) -> None:
|
|
snapshot_dir = _snapshot(tmp_path)
|
|
now = datetime(2026, 10, 16, 10, 0, 0).astimezone()
|
|
with pytest.raises(ValueError, match="finished_at"):
|
|
finish_snapshot(snapshot_dir, now, {"finished_at": "x"})
|
|
assert not (snapshot_dir / "finished.json").exists()
|
|
finish_snapshot(snapshot_dir, now, {})
|
|
first = (snapshot_dir / "finished.json").read_bytes()
|
|
with pytest.raises(FileExistsError):
|
|
finish_snapshot(snapshot_dir, now, {"segment_cache_hits": 1})
|
|
assert (snapshot_dir / "finished.json").read_bytes() == first
|