1af442481e
将后端迁入 backend/，完善根目录 .gitignore，删除误提交的 .mypy_cache 缓存文件。 Co-authored-by: Cursor <cursoragent@cursor.com>
231 lines
7.4 KiB
Python
231 lines
7.4 KiB
Python
"""consumption_log.txt 兼容 TSV 格式。"""
|
||
|
||
import pytest
|
||
|
||
from app.baked import pipeline as bp
|
||
from app.domain.vision_prediction import ClsTop3
|
||
from app.services.consumption_tsv_log import (
|
||
HEADER,
|
||
SUMMARY_HEADER,
|
||
_RANGE_SEP,
|
||
append_consumption_log_summary,
|
||
append_consumption_tsv_line,
|
||
build_consumption_markdown,
|
||
build_pending_consumption_markdown,
|
||
build_tsv_line,
|
||
format_pending_consumption_item_display_name,
|
||
init_consumption_log_file,
|
||
replace_pending_line_with_voice_resolution,
|
||
resolve_consumption_item_id,
|
||
short_camera_label,
|
||
)
|
||
|
||
|
||
def test_short_camera_label() -> None:
|
||
assert short_camera_label("or-cam-01") == "cam01"
|
||
assert short_camera_label("or-cam-2") == "cam02"
|
||
|
||
|
||
def test_build_tsv_line_matches_sample_shape(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setattr(bp, "CONSUMPTION_LOG_TIMEZONE", "UTC")
|
||
best = ClsTop3(
|
||
t1_name="一次性医用灭菌棉签",
|
||
t1_conf=0.9997,
|
||
t2_name="cls2",
|
||
t2_conf=0.0003,
|
||
t3_name="cls3",
|
||
t3_conf=0.0002,
|
||
t1_pid="2237844",
|
||
t2_pid="11765-1-101",
|
||
t3_pid="21504-1-1",
|
||
)
|
||
# 墙钟：拉流起点对齐到 2024-01-01T00:00:00Z，时间窗 +0s…+45s
|
||
w0 = 1704067200.0
|
||
line = build_tsv_line(
|
||
name_to_code={},
|
||
best=best,
|
||
doctor_id="DOCTOR_PLACEHOLDER",
|
||
camera_id="or-cam-01",
|
||
wall_start_epoch=w0,
|
||
wall_end_epoch=w0 + 45.0,
|
||
)
|
||
parts = line.rstrip("\n").split("\t")
|
||
assert len(parts) == 11
|
||
assert parts[0] == "2237844"
|
||
assert parts[1] == "一次性医用灭菌棉签"
|
||
assert parts[2] == "1"
|
||
assert parts[3] == "DOCTOR_PLACEHOLDER"
|
||
assert parts[4] == "cam01@2024-01-01T00:00:00.000+00:00" + _RANGE_SEP + "2024-01-01T00:00:45.000+00:00"
|
||
assert parts[5] == "一次性医用灭菌棉签"
|
||
assert parts[6] == "0.9997"
|
||
assert parts[7] == "cls2"
|
||
assert parts[8] == "0.0003"
|
||
assert parts[9] == "cls3"
|
||
assert parts[10] == "0.0002"
|
||
|
||
|
||
def test_resolve_consumption_item_id_uses_normalized_catalog_key() -> None:
|
||
name_to_code = {"一次性使用手术单（一次性医用垫单）": "PID-900"}
|
||
assert resolve_consumption_item_id("一次性医用垫单", "", name_to_code) == "PID-900"
|
||
|
||
|
||
def test_header_columns() -> None:
|
||
cols = HEADER.strip().split("\t")
|
||
assert cols == [
|
||
"item_id",
|
||
"item_name",
|
||
"qty",
|
||
"doctor_id",
|
||
"timestamp",
|
||
"top1_name",
|
||
"top1_conf",
|
||
"top2_name",
|
||
"top2_conf",
|
||
"top3_name",
|
||
"top3_conf",
|
||
]
|
||
|
||
|
||
def test_replace_pending_line_with_voice_resolution_rewrites_one_row(
|
||
tmp_path: object,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""语音确认后应替换 pending 行，而不是再多一行。"""
|
||
monkeypatch.setattr(bp, "CONSUMPTION_TSV_LOG_ENABLED", True)
|
||
monkeypatch.setattr(bp, "CONSUMPTION_LOG_TIMEZONE", "UTC")
|
||
monkeypatch.setattr(
|
||
bp,
|
||
"CONSUMPTION_TSV_LOG_PATH",
|
||
str(tmp_path / "{surgery_id}.txt"),
|
||
)
|
||
init_consumption_log_file("SURG01")
|
||
pending = (
|
||
"pending:abc-123\t一次性针头（待确认）\t1\tvision\t"
|
||
"cam01@2024-01-01T00:00:00.000+00:00"
|
||
f"{_RANGE_SEP}2024-01-01T00:00:45.000+00:00\t"
|
||
"一次性针头\t0.5000\tx\t0.1\ty\t0.2\n"
|
||
)
|
||
append_consumption_tsv_line("SURG01", pending)
|
||
replace_pending_line_with_voice_resolution(
|
||
surgery_id="SURG01",
|
||
confirmation_id="abc-123",
|
||
name_to_code={"纱布": "G1"},
|
||
chosen_label="纱布",
|
||
doctor_id="voice",
|
||
wall_epoch=1704067200.0,
|
||
)
|
||
text = (tmp_path / "SURG01.txt").read_text(encoding="utf-8")
|
||
assert "一次性针头（待确认）" not in text
|
||
assert "pending:abc-123" not in text
|
||
assert "纱布" in text
|
||
assert "G1" in text
|
||
# HEADER + 恰好一行数据
|
||
data_lines = [ln for ln in text.splitlines() if ln and not ln.startswith("item_id\t")]
|
||
assert len(data_lines) == 1
|
||
|
||
|
||
def test_per_surgery_file_init_and_append(
|
||
tmp_path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
monkeypatch.setattr(bp, "CONSUMPTION_TSV_LOG_ENABLED", True)
|
||
monkeypatch.setattr(
|
||
bp,
|
||
"CONSUMPTION_TSV_LOG_PATH",
|
||
str(tmp_path / "{surgery_id}.txt"),
|
||
)
|
||
init_consumption_log_file("or-001")
|
||
append_consumption_tsv_line("or-001", "row1\n")
|
||
append_consumption_tsv_line("or-001", "row2\n")
|
||
p = tmp_path / "or-001.txt"
|
||
assert p.read_text(encoding="utf-8") == HEADER + "row1\n" + "row2\n"
|
||
init_consumption_log_file("or-001")
|
||
assert p.read_text(encoding="utf-8") == HEADER
|
||
|
||
|
||
def test_append_consumption_log_summary_appends_three_column_block(
|
||
tmp_path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
monkeypatch.setattr(bp, "CONSUMPTION_TSV_LOG_ENABLED", True)
|
||
monkeypatch.setattr(
|
||
bp,
|
||
"CONSUMPTION_TSV_LOG_PATH",
|
||
str(tmp_path / "{surgery_id}.txt"),
|
||
)
|
||
init_consumption_log_file("s1")
|
||
append_consumption_tsv_line("s1", "x\n")
|
||
append_consumption_log_summary(
|
||
"s1",
|
||
{"A": ("nA", 2), "B": ("nB", 1)},
|
||
)
|
||
text = (tmp_path / "s1.txt").read_text(encoding="utf-8")
|
||
assert text.endswith("\n" + SUMMARY_HEADER + "A\tnA\t2\n" + "B\tnB\t1\n")
|
||
|
||
|
||
def test_build_consumption_markdown_top123_columns(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setattr(bp, "CONSUMPTION_LOG_TIMEZONE", "UTC")
|
||
best = ClsTop3(
|
||
t1_name="一次性医用灭菌棉签",
|
||
t1_conf=0.9997,
|
||
t2_name="cls2",
|
||
t2_conf=0.0003,
|
||
t3_name="cls3",
|
||
t3_conf=0.0002,
|
||
t1_pid="2237844",
|
||
t2_pid="11765-1-101",
|
||
t3_pid="21504-1-1",
|
||
)
|
||
w0 = 1704067200.0
|
||
md = build_consumption_markdown(
|
||
name_to_code={},
|
||
best=best,
|
||
doctor_id="DOCTOR_PLACEHOLDER",
|
||
camera_id="or-cam-01",
|
||
wall_start_epoch=w0,
|
||
wall_end_epoch=w0 + 45.0,
|
||
)
|
||
assert "| item_id |" in md and "| item_name |" in md and "| qty |" in md
|
||
assert "| 相对开录 |" in md
|
||
assert "| top2 |" in md and "| top3 |" in md
|
||
assert "2237844" in md
|
||
assert "一次性医用灭菌棉签" in md
|
||
assert "cls2" in md and "cls3" in md
|
||
assert "DOCTOR_PLACEHOLDER" in md
|
||
assert "| 1 |" in md
|
||
# 终端为可读时间戳，非落盘用 ISO@cam
|
||
assert "2024-01-01 00:00:00.000" in md and "2024-01-01 00:00:45.000" in md
|
||
assert "cam01" in md and " · " in md and _RANGE_SEP in md
|
||
assert "cam01@2024-01" not in md
|
||
assert "| DOCTOR_PLACEHOLDER | — |" in md
|
||
|
||
|
||
def test_build_pending_consumption_markdown_top123(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setattr(bp, "CONSUMPTION_LOG_TIMEZONE", "UTC")
|
||
snap = ClsTop3(
|
||
t1_name="输液器A",
|
||
t1_conf=0.88,
|
||
t2_name="候选二",
|
||
t2_conf=0.07,
|
||
t3_name="候选三",
|
||
t3_conf=0.02,
|
||
t1_pid="",
|
||
t2_pid="",
|
||
t3_pid="",
|
||
)
|
||
w0 = 1704067200.0
|
||
md = build_pending_consumption_markdown(
|
||
confirmation_id="cid-1",
|
||
item_display_name=format_pending_consumption_item_display_name(model_top1_label="输液器A"),
|
||
model_snap=snap,
|
||
doctor_id="vision",
|
||
camera_id="or-cam-01",
|
||
wall_start_epoch=w0,
|
||
wall_end_epoch=w0 + 15.0,
|
||
since_recording_start="0分15秒",
|
||
)
|
||
assert "| top1 | top2 | top3 |" in md
|
||
assert "输液器A (0.8800)" in md
|
||
assert "候选二 (0.0700)" in md
|
||
assert "候选三 (0.0200)" in md
|