admin/PACS
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Refine PACS viewer annotation workflow

Browse Source
This commit is contained in:
Codex
2026-05-27 10:25:33 +08:00
parent 9c478ed392
commit ac75c37e3f
5 changed files with 971 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files

524
PACS_DICOM处理/数据处理网页端/app.py
View File

@@ -76,7 +76,7 @@ app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
TOKENS: dict[str, str] = {}
STUDY_CACHE: dict[str, dict[str, Any]] = {}
STACK_CACHE: dict[str, tuple[float, np.ndarray]] = {}
STACK_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
class LoginIn(BaseModel):
@@ -86,9 +86,17 @@ class LoginIn(BaseModel):
class AnnotationIn(BaseModel):
body_parts: list[str] = []
manual_body_parts: list[str] = []
ai_body_parts: list[str] = []
upper_abdomen_phase: str = ""
manual_upper_abdomen_phase: str = ""
ai_upper_abdomen_phase: str = ""
plain_ct: bool = False
manual_plain_ct: bool | None = None
ai_plain_ct: bool | None = None
notes: str = ""
skipped: bool = False
ai_skipped: bool = False
class AIRequest(BaseModel):
@@ -174,8 +182,16 @@ def ensure_annotation_table() -> None:
series_instance_uid text NOT NULL,
series_description text,
body_parts jsonb NOT NULL DEFAULT '[]'::jsonb,
manual_body_parts jsonb NOT NULL DEFAULT '[]'::jsonb,
ai_body_parts jsonb NOT NULL DEFAULT '[]'::jsonb,
upper_abdomen_phase text NOT NULL DEFAULT '',
manual_upper_abdomen_phase text NOT NULL DEFAULT '',
ai_upper_abdomen_phase text NOT NULL DEFAULT '',
plain_ct boolean NOT NULL DEFAULT false,
manual_plain_ct boolean,
ai_plain_ct boolean,
skipped boolean NOT NULL DEFAULT false,
ai_skipped boolean NOT NULL DEFAULT false,
notes text NOT NULL DEFAULT '',
ai_result jsonb,
ai_model text,
@@ -184,12 +200,34 @@ def ensure_annotation_table() -> None:
updated_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (ct_number, series_instance_uid)
);
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS manual_body_parts jsonb NOT NULL DEFAULT '[]'::jsonb;
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS ai_body_parts jsonb NOT NULL DEFAULT '[]'::jsonb;
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS manual_upper_abdomen_phase text NOT NULL DEFAULT '';
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS ai_upper_abdomen_phase text NOT NULL DEFAULT '';
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS plain_ct boolean NOT NULL DEFAULT false;
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS manual_plain_ct boolean;
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS ai_plain_ct boolean;
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS skipped boolean NOT NULL DEFAULT false;
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS ai_skipped boolean NOT NULL DEFAULT false;
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS ai_result jsonb;
ALTER TABLE public.pacs_dicom_series_annotations
ADD COLUMN IF NOT EXISTS ai_model text;
UPDATE public.pacs_dicom_series_annotations
SET manual_body_parts = body_parts,
manual_upper_abdomen_phase = upper_abdomen_phase
WHERE jsonb_array_length(body_parts) > 0
AND jsonb_array_length(manual_body_parts) = 0
AND jsonb_array_length(ai_body_parts) = 0;
"""
pg_scalar(sql)
@@ -372,6 +410,13 @@ def read_header(path: Path) -> dict[str, str]:
tags = [
"SeriesInstanceUID",
"StudyInstanceUID",
"AccessionNumber",
"PatientName",
"PatientID",
"PatientBirthDate",
"PatientSex",
"InstitutionName",
"StudyDate",
"SeriesNumber",
"SeriesDescription",
"InstanceNumber",
@@ -414,11 +459,101 @@ def sort_key(item: tuple[Path, dict[str, str]]) -> tuple[float, float, str]:
return (z, instance, str(path))
def valid_parts(parts: Any) -> list[str]:
if not isinstance(parts, list):
return []
return [part for part in parts if part in BODY_PARTS]
def valid_phase(value: Any) -> str:
return value if value in PHASES else ""
def bool_or_none(value: Any) -> bool | None:
if value is None:
return None
if isinstance(value, bool):
return value
if isinstance(value, str):
if value.lower() in {"true", "t", "1", "yes", "y"}:
return True
if value.lower() in {"false", "f", "0", "no", "n"}:
return False
return bool(value)
def merge_parts(manual_parts: list[str], ai_parts: list[str], skipped: bool) -> list[str]:
if skipped:
return []
merged: list[str] = []
for part in manual_parts + ai_parts:
if part in BODY_PARTS and part not in merged:
merged.append(part)
return merged
def effective_phase(manual_phase: str, ai_phase: str, body_parts: list[str], skipped: bool) -> str:
if skipped or "upper_abdomen" not in body_parts:
return ""
return valid_phase(manual_phase) or valid_phase(ai_phase)
def effective_plain_ct(manual_plain_ct: bool | None, ai_plain_ct: bool | None, fallback: bool, skipped: bool) -> bool:
if skipped:
return False
if manual_plain_ct is not None:
return manual_plain_ct
if ai_plain_ct is not None:
return ai_plain_ct
return bool(fallback)
def normalize_annotation(row: dict[str, Any] | None, default_skipped: bool = False) -> dict[str, Any]:
row = row or {}
manual_parts = valid_parts(row.get("manual_body_parts") or [])
ai_parts = valid_parts(row.get("ai_body_parts") or [])
if not manual_parts and not ai_parts:
manual_parts = valid_parts(row.get("body_parts") or [])
skipped = bool(row.get("skipped", default_skipped))
ai_skipped = bool(row.get("ai_skipped", False))
manual_phase = valid_phase(row.get("manual_upper_abdomen_phase") or "")
ai_phase = valid_phase(row.get("ai_upper_abdomen_phase") or "")
if not manual_phase and not ai_phase:
manual_phase = valid_phase(row.get("upper_abdomen_phase") or "")
body_parts = merge_parts(manual_parts, ai_parts, skipped)
plain_ct = effective_plain_ct(
bool_or_none(row.get("manual_plain_ct")),
bool_or_none(row.get("ai_plain_ct")),
bool(row.get("plain_ct", False)),
skipped,
)
return {
"body_parts": body_parts,
"manual_body_parts": [] if skipped else manual_parts,
"ai_body_parts": [] if skipped else ai_parts,
"upper_abdomen_phase": effective_phase(manual_phase, ai_phase, body_parts, skipped),
"manual_upper_abdomen_phase": "" if skipped else manual_phase,
"ai_upper_abdomen_phase": "" if skipped else ai_phase,
"plain_ct": plain_ct,
"manual_plain_ct": None if skipped else bool_or_none(row.get("manual_plain_ct")),
"ai_plain_ct": None if skipped else bool_or_none(row.get("ai_plain_ct")),
"skipped": skipped,
"ai_skipped": ai_skipped,
"notes": row.get("notes", ""),
"updated_at": row.get("updated_at", ""),
"ai_model": row.get("ai_model", ""),
}
def get_annotations(ct_number: str) -> dict[str, dict[str, Any]]:
try:
ensure_annotation_table()
rows = pg_json_rows(
f"""
SELECT series_instance_uid, body_parts, upper_abdomen_phase, skipped, notes, updated_at, ai_model
SELECT
series_instance_uid, body_parts, manual_body_parts, ai_body_parts,
upper_abdomen_phase, manual_upper_abdomen_phase, ai_upper_abdomen_phase,
plain_ct, manual_plain_ct, ai_plain_ct, skipped, ai_skipped, notes, updated_at, ai_model
FROM public.pacs_dicom_series_annotations
WHERE ct_number = {sql_literal(ct_number)}
"""
@@ -450,17 +585,35 @@ def scan_study(ct_number: str) -> dict[str, Any]:
annotations = get_annotations(ct_number)
series_list = []
file_map = {}
default_skip_rows = []
for uid, items in grouped.items():
items.sort(key=sort_key)
first = items[0][1]
last = items[-1][1]
file_map[uid] = [path for path, _ in items]
annotation = annotations.get(uid, {})
raw_annotation = annotations.get(uid)
annotation = normalize_annotation(raw_annotation)
default_skipped = len(items) < 80 and not annotation.get("body_parts") and not annotation.get("plain_ct")
if default_skipped:
annotation["skipped"] = True
annotation["manual_body_parts"] = []
annotation["ai_body_parts"] = []
annotation["body_parts"] = []
annotation["plain_ct"] = False
if default_skipped:
default_skip_rows.append((uid, first, len(items)))
series_list.append(
{
"ct_number": ct_number,
"series_uid": uid,
"study_uid": first.get("StudyInstanceUID", ""),
"accession_number": first.get("AccessionNumber", ""),
"patient_name": first.get("PatientName", ""),
"patient_id": first.get("PatientID", ""),
"patient_birth_date": first.get("PatientBirthDate", ""),
"patient_sex": first.get("PatientSex", ""),
"institution_name": first.get("InstitutionName", ""),
"study_date": first.get("StudyDate", "") or study.get("study_date", ""),
"series_number": first.get("SeriesNumber", ""),
"description": first.get("SeriesDescription", "") or "未命名序列",
"count": len(items),
@@ -476,24 +629,75 @@ def scan_study(ct_number: str) -> dict[str, Any]:
"pixel_spacing": first.get("PixelSpacing", ""),
"slice_thickness": first.get("SliceThickness", ""),
"spacing_between_slices": first.get("SpacingBetweenSlices", ""),
"annotation": {
"body_parts": annotation.get("body_parts", []),
"upper_abdomen_phase": annotation.get("upper_abdomen_phase", ""),
"skipped": bool(annotation.get("skipped", False)),
"notes": annotation.get("notes", ""),
"updated_at": annotation.get("updated_at", ""),
"ai_model": annotation.get("ai_model", ""),
},
"window_center": first.get("WindowCenter", ""),
"window_width": first.get("WindowWidth", ""),
"default_skipped": default_skipped,
"annotation": annotation,
}
)
for uid, first, count in default_skip_rows:
try:
pg_scalar(
f"""
INSERT INTO public.pacs_dicom_series_annotations (
ct_number, study_instance_uid, series_instance_uid, series_description,
body_parts, manual_body_parts, ai_body_parts, upper_abdomen_phase,
plain_ct, skipped, notes, updated_by, updated_at
)
VALUES (
{sql_literal(ct_number)},
{sql_literal(first.get('StudyInstanceUID', ''))},
{sql_literal(uid)},
{sql_literal(first.get('SeriesDescription', '') or '未命名序列')},
'[]'::jsonb,
'[]'::jsonb,
'[]'::jsonb,
'',
false,
true,
{sql_literal(f'少于80张默认略过（{count}张）')},
'system',
now()
)
ON CONFLICT (ct_number, series_instance_uid) DO UPDATE SET
skipped = true,
body_parts = '[]'::jsonb,
manual_body_parts = '[]'::jsonb,
ai_body_parts = '[]'::jsonb,
upper_abdomen_phase = '',
manual_upper_abdomen_phase = '',
ai_upper_abdomen_phase = '',
plain_ct = false,
notes = CASE
WHEN pacs_dicom_series_annotations.notes = ''
THEN EXCLUDED.notes
ELSE pacs_dicom_series_annotations.notes
END,
updated_by = 'system',
updated_at = now()
WHERE jsonb_array_length(pacs_dicom_series_annotations.body_parts) = 0
AND pacs_dicom_series_annotations.plain_ct IS NOT TRUE
""",
timeout=8,
)
except Exception:
pass
def series_time_key(row: dict[str, Any]) -> tuple[str, int, str]:
return (
row.get("series_time") or row.get("first_time") or row.get("study_time") or "",
numeric(row.get("series_number", "")),
row.get("description", ""),
)
def numeric(value: str) -> int:
try:
return int(float(value))
except Exception:
return 999999
series_list.sort(key=lambda row: (1 if row["annotation"].get("skipped") else 0, numeric(row["series_number"]), row["description"]))
series_list.sort(key=lambda row: (1 if row["annotation"].get("skipped") else 0, *series_time_key(row)))
cached = {"cached_at": time.time(), "study": study, "series": series_list, "files": file_map}
STUDY_CACHE[ct_number] = cached
return cached
@@ -535,13 +739,91 @@ def dicom_to_hu(ds: pydicom.Dataset) -> np.ndarray:
return arr * slope + intercept
def render_array(arr: np.ndarray, center: float, width: float, invert: bool = False, rotate: int = 0, max_size: int = 900) -> bytes:
def as_float(value: Any, default: float = 1.0) -> float:
try:
return float(value)
except Exception:
return default
def pixel_spacing_from_ds(ds: pydicom.Dataset) -> tuple[float, float]:
spacing = getattr(ds, "PixelSpacing", None)
if spacing and len(spacing) >= 2:
row_spacing = as_float(spacing[0], 1.0)
col_spacing = as_float(spacing[1], row_spacing)
return max(row_spacing, 0.001), max(col_spacing, 0.001)
return 1.0, 1.0
def slice_spacing_from_datasets(datasets: list[pydicom.Dataset]) -> float:
distances = []
positions = []
for ds in datasets:
position = getattr(ds, "ImagePositionPatient", None)
if position and len(position) >= 3:
positions.append(np.array([as_float(position[0]), as_float(position[1]), as_float(position[2])], dtype=np.float32))
for first, second in zip(positions, positions[1:]):
distance = float(np.linalg.norm(second - first))
if distance > 0.001:
distances.append(distance)
if distances:
return max(float(np.median(distances)), 0.001)
locations = []
for ds in datasets:
value = getattr(ds, "SliceLocation", None)
if value is not None:
locations.append(as_float(value, 0.0))
for first, second in zip(locations, locations[1:]):
distance = abs(second - first)
if distance > 0.001:
distances.append(distance)
if distances:
return max(float(np.median(distances)), 0.001)
sample = datasets[0] if datasets else None
if sample is not None:
spacing = as_float(getattr(sample, "SpacingBetweenSlices", 0), 0.0)
if spacing > 0.001:
return spacing
thickness = as_float(getattr(sample, "SliceThickness", 0), 0.0)
if thickness > 0.001:
return thickness
return 1.0
def resize_for_spacing(pil: Image.Image, row_spacing: float, col_spacing: float) -> Image.Image:
row_spacing = max(row_spacing, 0.001)
col_spacing = max(col_spacing, 0.001)
if abs(row_spacing - col_spacing) < 0.01:
return pil
base = min(row_spacing, col_spacing)
target_w = max(1, int(round(pil.width * col_spacing / base)))
target_h = max(1, int(round(pil.height * row_spacing / base)))
max_edge = 2200
scale = min(1.0, max_edge / max(target_w, target_h))
target_size = (max(1, int(round(target_w * scale))), max(1, int(round(target_h * scale))))
if target_size == pil.size:
return pil
return pil.resize(target_size, Image.Resampling.BILINEAR)
def render_array(
arr: np.ndarray,
center: float,
width: float,
invert: bool = False,
rotate: int = 0,
max_size: int = 900,
pixel_spacing: tuple[float, float] = (1.0, 1.0),
) -> bytes:
low = center - width / 2.0
high = center + width / 2.0
img = ((np.clip(arr, low, high) - low) / max(high - low, 1.0) * 255.0).astype(np.uint8)
if invert:
img = 255 - img
pil = Image.fromarray(img)
pil = resize_for_spacing(pil, pixel_spacing[0], pixel_spacing[1])
if rotate:
pil = pil.rotate(-rotate, expand=True)
if max(pil.size) > max_size:
@@ -551,7 +833,7 @@ def render_array(arr: np.ndarray, center: float, width: float, invert: bool = Fa
return output.getvalue()
def load_stack(ct_number: str, series_uid: str) -> np.ndarray:
def load_stack_data(ct_number: str, series_uid: str) -> dict[str, Any]:
key = f"{ct_number}|{series_uid}"
cached = STACK_CACHE.get(key)
if cached:
@@ -559,15 +841,24 @@ def load_stack(ct_number: str, series_uid: str) -> np.ndarray:
return cached[1]
files = get_series_files(ct_number, series_uid)
arrays = []
datasets = []
for path in files:
ds = pydicom.dcmread(str(path), force=True)
datasets.append(ds)
arrays.append(dicom_to_hu(ds))
stack = np.stack(arrays, axis=0)
STACK_CACHE[key] = (time.time(), stack)
row_spacing, col_spacing = pixel_spacing_from_ds(datasets[min(len(datasets) - 1, len(datasets) // 2)])
payload = {
"stack": stack,
"row_spacing": row_spacing,
"col_spacing": col_spacing,
"slice_spacing": slice_spacing_from_datasets(datasets),
}
STACK_CACHE[key] = (time.time(), payload)
if len(STACK_CACHE) > 2:
oldest = sorted(STACK_CACHE.items(), key=lambda item: item[1][0])[0][0]
STACK_CACHE.pop(oldest, None)
return stack
return payload
@app.get("/api/image")
@@ -586,21 +877,31 @@ def image(
index = min(index, len(files) - 1)
ds = pydicom.dcmread(str(files[index]), force=True)
center, width = window_values(ds, window)
payload = render_array(dicom_to_hu(ds), center, width, getattr(ds, "PhotometricInterpretation", "") == "MONOCHROME1", rotate)
payload = render_array(
dicom_to_hu(ds),
center,
width,
getattr(ds, "PhotometricInterpretation", "") == "MONOCHROME1",
rotate,
pixel_spacing=pixel_spacing_from_ds(ds),
)
return Response(payload, media_type="image/png")
stack = load_stack(ct_number, series_uid)
stack_data = load_stack_data(ct_number, series_uid)
stack = stack_data["stack"]
sample_ds = pydicom.dcmread(str(files[min(len(files) - 1, len(files) // 2)]), stop_before_pixels=True, force=True)
center, width = window_values(sample_ds, window)
if plane == "coronal":
index = min(index, stack.shape[1] - 1)
arr = stack[:, index, :]
spacing = (stack_data["slice_spacing"], stack_data["col_spacing"])
elif plane == "sagittal":
index = min(index, stack.shape[2] - 1)
arr = stack[:, :, index]
spacing = (stack_data["slice_spacing"], stack_data["row_spacing"])
else:
raise HTTPException(status_code=400, detail="invalid plane")
payload = render_array(np.flipud(arr), center, width, False, rotate)
payload = render_array(np.flipud(arr), center, width, False, rotate, pixel_spacing=spacing)
return Response(payload, media_type="image/png")
@@ -644,22 +945,45 @@ def dicom_info(ct_number: str, series_uid: str, index: int = 0, _: str = Depends
return {"path": str(path), "fields": fields}
@app.put("/api/series/{ct_number}/{series_uid}/annotation")
def save_annotation(ct_number: str, series_uid: str, data: AnnotationIn, user: str = Depends(require_auth)) -> dict[str, Any]:
body_parts = [] if data.skipped else [part for part in data.body_parts if part in BODY_PARTS]
phase = data.upper_abdomen_phase if data.upper_abdomen_phase in PHASES else ""
if "upper_abdomen" not in body_parts:
phase = ""
study = scan_study(ct_number)
series_row = next((row for row in study["series"] if row["series_uid"] == series_uid), None)
if not series_row:
raise HTTPException(status_code=404, detail="series not found")
def sql_bool_or_null(value: bool | None) -> str:
if value is None:
return "NULL"
return "true" if value else "false"
def save_annotation_payload(
ct_number: str,
series_uid: str,
series_row: dict[str, Any],
manual_parts: list[str],
ai_parts: list[str],
manual_phase: str,
ai_phase: str,
manual_plain_ct: bool | None,
ai_plain_ct: bool | None,
skipped: bool,
ai_skipped: bool,
notes: str,
user: str,
ai_result: dict[str, Any] | None = None,
ai_model: str | None = None,
) -> dict[str, Any]:
manual_parts = valid_parts(manual_parts)
ai_parts = valid_parts(ai_parts)
body_parts = merge_parts(manual_parts, ai_parts, skipped)
manual_phase = valid_phase(manual_phase)
ai_phase = valid_phase(ai_phase)
phase = effective_phase(manual_phase, ai_phase, body_parts, skipped)
plain_ct = effective_plain_ct(manual_plain_ct, ai_plain_ct, False, skipped)
ensure_annotation_table()
pg_scalar(
f"""
INSERT INTO public.pacs_dicom_series_annotations (
ct_number, study_instance_uid, series_instance_uid, series_description,
body_parts, upper_abdomen_phase, skipped, notes, updated_by, updated_at
body_parts, manual_body_parts, ai_body_parts,
upper_abdomen_phase, manual_upper_abdomen_phase, ai_upper_abdomen_phase,
plain_ct, manual_plain_ct, ai_plain_ct,
skipped, ai_skipped, notes, ai_result, ai_model, updated_by, updated_at
)
VALUES (
{sql_literal(ct_number)},
@@ -667,9 +991,19 @@ def save_annotation(ct_number: str, series_uid: str, data: AnnotationIn, user: s
{sql_literal(series_uid)},
{sql_literal(series_row.get('description', ''))},
{sql_literal(json.dumps(body_parts, ensure_ascii=False))}::jsonb,
{sql_literal(json.dumps([] if skipped else manual_parts, ensure_ascii=False))}::jsonb,
{sql_literal(json.dumps([] if skipped else ai_parts, ensure_ascii=False))}::jsonb,
{sql_literal(phase)},
{'true' if data.skipped else 'false'},
{sql_literal(data.notes)},
{sql_literal('' if skipped else manual_phase)},
{sql_literal('' if skipped else ai_phase)},
{'true' if plain_ct else 'false'},
{sql_bool_or_null(None if skipped else manual_plain_ct)},
{sql_bool_or_null(None if skipped else ai_plain_ct)},
{'true' if skipped else 'false'},
{'true' if ai_skipped else 'false'},
{sql_literal(notes)},
{sql_literal(json.dumps(ai_result, ensure_ascii=False)) + '::jsonb' if ai_result is not None else 'NULL'},
{sql_literal(ai_model) if ai_model is not None else 'NULL'},
{sql_literal(user)},
now()
)
@@ -677,15 +1011,65 @@ def save_annotation(ct_number: str, series_uid: str, data: AnnotationIn, user: s
study_instance_uid = EXCLUDED.study_instance_uid,
series_description = EXCLUDED.series_description,
body_parts = EXCLUDED.body_parts,
manual_body_parts = EXCLUDED.manual_body_parts,
ai_body_parts = EXCLUDED.ai_body_parts,
upper_abdomen_phase = EXCLUDED.upper_abdomen_phase,
manual_upper_abdomen_phase = EXCLUDED.manual_upper_abdomen_phase,
ai_upper_abdomen_phase = EXCLUDED.ai_upper_abdomen_phase,
plain_ct = EXCLUDED.plain_ct,
manual_plain_ct = EXCLUDED.manual_plain_ct,
ai_plain_ct = EXCLUDED.ai_plain_ct,
skipped = EXCLUDED.skipped,
ai_skipped = EXCLUDED.ai_skipped,
notes = EXCLUDED.notes,
ai_result = COALESCE(EXCLUDED.ai_result, pacs_dicom_series_annotations.ai_result),
ai_model = COALESCE(EXCLUDED.ai_model, pacs_dicom_series_annotations.ai_model),
updated_by = EXCLUDED.updated_by,
updated_at = now()
"""
)
return normalize_annotation(
{
"body_parts": body_parts,
"manual_body_parts": [] if skipped else manual_parts,
"ai_body_parts": [] if skipped else ai_parts,
"upper_abdomen_phase": phase,
"manual_upper_abdomen_phase": "" if skipped else manual_phase,
"ai_upper_abdomen_phase": "" if skipped else ai_phase,
"plain_ct": plain_ct,
"manual_plain_ct": None if skipped else manual_plain_ct,
"ai_plain_ct": None if skipped else ai_plain_ct,
"skipped": skipped,
"ai_skipped": ai_skipped,
"notes": notes,
"ai_model": ai_model or series_row.get("annotation", {}).get("ai_model", ""),
}
)
@app.put("/api/series/{ct_number}/{series_uid}/annotation")
def save_annotation(ct_number: str, series_uid: str, data: AnnotationIn, user: str = Depends(require_auth)) -> dict[str, Any]:
study = scan_study(ct_number)
series_row = next((row for row in study["series"] if row["series_uid"] == series_uid), None)
if not series_row:
raise HTTPException(status_code=404, detail="series not found")
annotation = save_annotation_payload(
ct_number=ct_number,
series_uid=series_uid,
series_row=series_row,
manual_parts=data.manual_body_parts or data.body_parts,
ai_parts=data.ai_body_parts,
manual_phase=data.manual_upper_abdomen_phase or data.upper_abdomen_phase,
ai_phase=data.ai_upper_abdomen_phase,
manual_plain_ct=data.manual_plain_ct if data.manual_plain_ct is not None else (data.plain_ct if data.plain_ct and data.ai_plain_ct is None else None),
ai_plain_ct=data.ai_plain_ct,
skipped=data.skipped,
ai_skipped=data.ai_skipped,
notes=data.notes,
user=user,
)
STUDY_CACHE.pop(ct_number, None)
return {"ok": True, "body_parts": body_parts, "upper_abdomen_phase": phase, "skipped": data.skipped}
return {"ok": True, **annotation}
def representative_images(ct_number: str, series_uid: str) -> list[tuple[str, bytes]]:
@@ -693,17 +1077,30 @@ def representative_images(ct_number: str, series_uid: str) -> list[tuple[str, by
axial_index = max(0, min(len(files) - 1, len(files) // 2))
ds = pydicom.dcmread(str(files[axial_index]), force=True)
center, width = window_values(ds, "soft")
images = [("横断面", render_array(dicom_to_hu(ds), center, width, getattr(ds, "PhotometricInterpretation", "") == "MONOCHROME1", max_size=720))]
images = [
(
"轴位原始",
render_array(
dicom_to_hu(ds),
center,
width,
getattr(ds, "PhotometricInterpretation", "") == "MONOCHROME1",
max_size=720,
pixel_spacing=pixel_spacing_from_ds(ds),
),
)
]
if len(files) >= 3:
try:
stack = load_stack(ct_number, series_uid)
stack_data = load_stack_data(ct_number, series_uid)
stack = stack_data["stack"]
sample_ds = pydicom.dcmread(str(files[axial_index]), stop_before_pixels=True, force=True)
center, width = window_values(sample_ds, "soft")
coronal = np.flipud(stack[:, stack.shape[1] // 2, :])
sagittal = np.flipud(stack[:, :, stack.shape[2] // 2])
images.append(("矢状面", render_array(sagittal, center, width, False, max_size=720)))
images.append(("冠状面", render_array(coronal, center, width, False, max_size=720)))
images.append(("矢状位重建", render_array(sagittal, center, width, False, max_size=720, pixel_spacing=(stack_data["slice_spacing"], stack_data["row_spacing"]))))
images.append(("冠状位重建", render_array(coronal, center, width, False, max_size=720, pixel_spacing=(stack_data["slice_spacing"], stack_data["col_spacing"]))))
except Exception:
pass
return images
@@ -742,8 +1139,9 @@ def ai_classify(ct_number: str, series_uid: str, _: AIRequest, user: str = Depen
"可选部位键: head_neck(头颈部), chest(胸部), upper_abdomen(上腹部), "
"lower_abdomen(下腹部), pelvis(盆腔)。一个序列可包含多个部位。"
"如果不是可用于标注的平扫CT影像、定位像、剂量报告或无法判断，请 skipped=true。"
"请同时判断 plain_ct 是否为平扫CT。"
"如果包含上腹部，请判断期相: arterial(动脉期)、portal_venous(门静脉期)、unknown(无法判别)。"
"只返回JSON: {\"body_parts\":[],\"upper_abdomen_phase\":\"\",\"skipped\":false,\"notes\":\"\"}。"
"只返回JSON: {\"body_parts\":[],\"upper_abdomen_phase\":\"\",\"plain_ct\":false,\"skipped\":false,\"notes\":\"\"}。"
f"PACS张数: {series_row.get('count', 0)}。"
f"序列描述: {series_row.get('description','')}；DICOM部位: {series_row.get('body_part_dicom','')}。"
),
@@ -781,36 +1179,27 @@ def ai_classify(ct_number: str, series_uid: str, _: AIRequest, user: str = Depen
suggestion = parse_ai_json(message)
body_parts = [part for part in suggestion.get("body_parts", []) if part in BODY_PARTS]
skipped = bool(suggestion.get("skipped", False))
plain_ct = bool(suggestion.get("plain_ct", False))
phase = suggestion.get("upper_abdomen_phase", "")
if phase not in PHASES or "upper_abdomen" not in body_parts:
phase = ""
ensure_annotation_table()
pg_scalar(
f"""
INSERT INTO public.pacs_dicom_series_annotations (
ct_number, study_instance_uid, series_instance_uid, series_description,
body_parts, upper_abdomen_phase, skipped, notes, ai_result, ai_model, updated_by, updated_at
)
VALUES (
{sql_literal(ct_number)},
{sql_literal(series_row.get('study_uid', ''))},
{sql_literal(series_uid)},
{sql_literal(series_row.get('description', ''))},
'[]'::jsonb,
'',
false,
'',
{sql_literal(json.dumps(suggestion, ensure_ascii=False))}::jsonb,
{sql_literal(KIMI_MODEL)},
{sql_literal(user)},
now()
)
ON CONFLICT (ct_number, series_instance_uid) DO UPDATE SET
ai_result = EXCLUDED.ai_result,
ai_model = EXCLUDED.ai_model,
updated_by = EXCLUDED.updated_by,
updated_at = now()
"""
current = normalize_annotation(series_row.get("annotation", {}))
annotation = save_annotation_payload(
ct_number=ct_number,
series_uid=series_uid,
series_row=series_row,
manual_parts=current.get("manual_body_parts", []),
ai_parts=[] if skipped else body_parts,
manual_phase=current.get("manual_upper_abdomen_phase", ""),
ai_phase="" if skipped else phase,
manual_plain_ct=current.get("manual_plain_ct"),
ai_plain_ct=None if skipped else plain_ct,
skipped=skipped,
ai_skipped=skipped,
notes=str(suggestion.get("notes", "")) or current.get("notes", ""),
user=user,
ai_result=suggestion,
ai_model=KIMI_MODEL,
)
STUDY_CACHE.pop(ct_number, None)
return {
@@ -818,10 +1207,7 @@ def ai_classify(ct_number: str, series_uid: str, _: AIRequest, user: str = Depen
"provider": "Kimi",
"name": KIMI_API_NAME,
"model": KIMI_MODEL,
"body_parts": [] if skipped else body_parts,
"upper_abdomen_phase": "" if skipped else phase,
"skipped": skipped,
"notes": str(suggestion.get("notes", "")),
**annotation,
"raw": suggestion,
}