Files
operating-room-monitor-server/app/services/tear_gated_segment_consumption/segments.py

79 lines
2.1 KiB
Python
Raw Normal View History

"""撕段时间线合并(与离线 demo 一致)。"""
from __future__ import annotations
from collections import Counter
def merge_tear_segments(
rows: list[tuple[int, float, bool, str]],
min_tear_sec: float = 3.0,
min_gap_sec: float = 1.5,
) -> list[dict]:
"""
rows: (frame_idx_in_clip, abs_time_sec, is_tear, consumable_label_or_empty)
"""
raw_segs: list[dict] = []
cur: dict | None = None
for frame_idx, t, is_tear, label in rows:
if is_tear:
if cur is None:
cur = {
"t0": t,
"t1": t,
"f0": frame_idx,
"f1": frame_idx,
"labels": [],
}
else:
gap = t - cur["t1"]
if gap > min_gap_sec:
raw_segs.append(cur)
cur = {
"t0": t,
"t1": t,
"f0": frame_idx,
"f1": frame_idx,
"labels": [],
}
cur["t1"] = t
cur["f1"] = frame_idx
if label:
cur["labels"].append(label)
else:
if cur is not None:
gap = t - cur["t1"]
if gap > min_gap_sec:
raw_segs.append(cur)
cur = None
if cur is not None:
raw_segs.append(cur)
valid: list[dict] = []
for s in raw_segs:
dur = s["t1"] - s["t0"]
if dur >= min_tear_sec - 1e-9:
valid.append(s)
out: list[dict] = []
for i, s in enumerate(valid, 1):
top_label = (
Counter(s["labels"]).most_common(1)[0][0]
if s["labels"]
else "(未识别耗材)"
)
out.append(
{
"index": i,
"start_sec": s["t0"],
"end_sec": s["t1"],
"duration_sec": s["t1"] - s["t0"],
"consumable": top_label,
"start_frame": s["f0"],
"end_frame": s["f1"],
}
)
return out