404 lines
14 KiB
Python
404 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Preprocess a PACS DICOM batch by normalizing study folders.
|
|
|
|
The top-level PACS export folder is assumed to contain one folder per study.
|
|
The exported folder name may contain the wrong CT number, so the canonical
|
|
ct_number is read from DICOM AccessionNumber (0008,0050).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
from collections import Counter, defaultdict
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
import pydicom
|
|
|
|
|
|
DICOM_TAGS = [
|
|
"AccessionNumber",
|
|
"PatientName",
|
|
"PatientID",
|
|
"PatientBirthDate",
|
|
"PatientSex",
|
|
"StudyInstanceUID",
|
|
"StudyDate",
|
|
"StudyTime",
|
|
"StudyID",
|
|
"Modality",
|
|
"BodyPartExamined",
|
|
"ProtocolName",
|
|
"StudyDescription",
|
|
"SeriesInstanceUID",
|
|
"SOPInstanceUID",
|
|
]
|
|
|
|
|
|
@dataclass
|
|
class StudyRow:
|
|
batch_name: str
|
|
source_folder_name: str
|
|
source_ct_number: str
|
|
source_patient_name: str
|
|
ct_number: str
|
|
target_folder_name: str
|
|
needs_ct_number_fix: bool
|
|
patient_name_dicom: str
|
|
patient_id: str
|
|
patient_birth_date: str
|
|
patient_sex: str
|
|
study_date: str
|
|
study_time: str
|
|
study_id: str
|
|
modality: str
|
|
body_part_examined: str
|
|
protocol_name: str
|
|
study_description: str
|
|
accession_numbers: str
|
|
raw_accession_numbers: str
|
|
study_instance_uids: str
|
|
series_count: int
|
|
dicom_file_count: int
|
|
total_file_count: int
|
|
total_bytes: int
|
|
source_path: str
|
|
processed_path: str
|
|
status: str
|
|
notes: str
|
|
|
|
|
|
def text(value: object) -> str:
|
|
if value is None:
|
|
return ""
|
|
return str(value).strip()
|
|
|
|
|
|
def most_common(counter: Counter[str]) -> str:
|
|
if not counter:
|
|
return ""
|
|
return counter.most_common(1)[0][0]
|
|
|
|
|
|
def sanitize_component(name: str) -> str:
|
|
cleaned = name.replace("/", "_").replace("\x00", "_").strip()
|
|
return cleaned or "UNKNOWN"
|
|
|
|
|
|
def parse_export_folder_name(name: str) -> tuple[str, str]:
|
|
if "-" not in name:
|
|
return name, ""
|
|
ct_number, patient_name = name.split("-", 1)
|
|
return ct_number.strip(), patient_name.strip()
|
|
|
|
|
|
def canonical_ct_number(accession_number: str) -> str:
|
|
accession_number = accession_number.strip()
|
|
match = re.fullmatch(r"((?:D)?CT\d+)-\d+", accession_number)
|
|
if match:
|
|
return match.group(1)
|
|
return accession_number
|
|
|
|
|
|
def iter_files(folder: Path) -> Iterable[Path]:
|
|
for path in folder.rglob("*"):
|
|
if path.is_file():
|
|
yield path
|
|
|
|
|
|
def read_meta(path: Path) -> dict[str, str]:
|
|
ds = pydicom.dcmread(
|
|
str(path),
|
|
stop_before_pixels=True,
|
|
force=True,
|
|
specific_tags=DICOM_TAGS + ["SpecificCharacterSet"],
|
|
)
|
|
return {tag: text(getattr(ds, tag, "")) for tag in DICOM_TAGS}
|
|
|
|
|
|
def scan_study_folder(batch_name: str, folder: Path, processed_batch_root: Path) -> tuple[StudyRow, list[dict[str, str]]]:
|
|
source_ct_number, source_patient_name = parse_export_folder_name(folder.name)
|
|
files = sorted(iter_files(folder))
|
|
|
|
counters: dict[str, Counter[str]] = defaultdict(Counter)
|
|
file_rows: list[dict[str, str]] = []
|
|
dicom_file_count = 0
|
|
total_bytes = 0
|
|
errors = []
|
|
|
|
for path in files:
|
|
try:
|
|
size = path.stat().st_size
|
|
total_bytes += size
|
|
meta = read_meta(path)
|
|
dicom_file_count += 1
|
|
except Exception as exc: # noqa: BLE001 - keep scanning and report the file.
|
|
errors.append(f"{path}: {exc}")
|
|
continue
|
|
|
|
for key in DICOM_TAGS:
|
|
value = meta.get(key, "")
|
|
if value:
|
|
counters[key][value] += 1
|
|
|
|
file_rows.append(
|
|
{
|
|
"ct_number": "",
|
|
"source_folder_name": folder.name,
|
|
"source_relative_path": str(path.relative_to(folder)),
|
|
"processed_relative_path": "",
|
|
"sop_instance_uid": meta.get("SOPInstanceUID", ""),
|
|
"series_instance_uid": meta.get("SeriesInstanceUID", ""),
|
|
"study_instance_uid": meta.get("StudyInstanceUID", ""),
|
|
"bytes": str(size),
|
|
}
|
|
)
|
|
|
|
raw_accession_numbers = sorted(counters["AccessionNumber"])
|
|
accession_numbers = sorted({canonical_ct_number(value) for value in raw_accession_numbers if value})
|
|
study_instance_uids = sorted(counters["StudyInstanceUID"])
|
|
|
|
status = "ok"
|
|
notes: list[str] = []
|
|
if errors:
|
|
status = "error"
|
|
notes.append(f"metadata_read_errors={len(errors)}")
|
|
if raw_accession_numbers and raw_accession_numbers != accession_numbers:
|
|
notes.append("normalized_accession_suffixes")
|
|
if not accession_numbers:
|
|
status = "error"
|
|
notes.append("missing_accession_number")
|
|
ct_number = source_ct_number
|
|
elif len(accession_numbers) > 1:
|
|
status = "error"
|
|
notes.append("multiple_accession_numbers")
|
|
ct_number = most_common(counters["AccessionNumber"])
|
|
else:
|
|
ct_number = accession_numbers[0]
|
|
|
|
target_folder_name = sanitize_component(ct_number)
|
|
if source_patient_name:
|
|
target_folder_name = f"{target_folder_name}-{sanitize_component(source_patient_name)}"
|
|
processed_path = processed_batch_root / target_folder_name
|
|
|
|
needs_ct_number_fix = source_ct_number != ct_number
|
|
|
|
for row in file_rows:
|
|
row["ct_number"] = ct_number
|
|
row["processed_relative_path"] = str(Path(target_folder_name) / row["source_relative_path"])
|
|
|
|
row = StudyRow(
|
|
batch_name=batch_name,
|
|
source_folder_name=folder.name,
|
|
source_ct_number=source_ct_number,
|
|
source_patient_name=source_patient_name,
|
|
ct_number=ct_number,
|
|
target_folder_name=target_folder_name,
|
|
needs_ct_number_fix=needs_ct_number_fix,
|
|
patient_name_dicom=most_common(counters["PatientName"]),
|
|
patient_id=most_common(counters["PatientID"]),
|
|
patient_birth_date=most_common(counters["PatientBirthDate"]),
|
|
patient_sex=most_common(counters["PatientSex"]),
|
|
study_date=most_common(counters["StudyDate"]),
|
|
study_time=most_common(counters["StudyTime"]),
|
|
study_id=most_common(counters["StudyID"]),
|
|
modality=most_common(counters["Modality"]),
|
|
body_part_examined=most_common(counters["BodyPartExamined"]),
|
|
protocol_name=most_common(counters["ProtocolName"]),
|
|
study_description=most_common(counters["StudyDescription"]),
|
|
accession_numbers="|".join(accession_numbers),
|
|
raw_accession_numbers="|".join(raw_accession_numbers),
|
|
study_instance_uids="|".join(study_instance_uids),
|
|
series_count=len(counters["SeriesInstanceUID"]),
|
|
dicom_file_count=dicom_file_count,
|
|
total_file_count=len(files),
|
|
total_bytes=total_bytes,
|
|
source_path=str(folder),
|
|
processed_path=str(processed_path),
|
|
status=status,
|
|
notes=";".join(notes),
|
|
)
|
|
return row, file_rows
|
|
|
|
|
|
def write_csv(path: Path, rows: list[dict[str, object]], fieldnames: list[str]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", newline="", encoding="utf-8") as fh:
|
|
writer = csv.DictWriter(fh, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
|
|
def hardlink_tree(source_folder: Path, target_folder: Path) -> tuple[int, int]:
|
|
linked = 0
|
|
already_ok = 0
|
|
for source in iter_files(source_folder):
|
|
rel = source.relative_to(source_folder)
|
|
target = target_folder / rel
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
if target.exists():
|
|
try:
|
|
if os.path.samefile(source, target):
|
|
already_ok += 1
|
|
continue
|
|
except OSError:
|
|
pass
|
|
if target.stat().st_size != source.stat().st_size:
|
|
raise RuntimeError(f"target exists with different size: {target}")
|
|
target.unlink()
|
|
try:
|
|
os.link(source, target)
|
|
except OSError as exc:
|
|
if exc.errno != 18: # EXDEV
|
|
raise
|
|
shutil.copy2(source, target)
|
|
linked += 1
|
|
return linked, already_ok
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--source", type=Path, required=True)
|
|
parser.add_argument("--processed-root", type=Path, required=True)
|
|
parser.add_argument("--results-root", type=Path, required=True)
|
|
parser.add_argument("--batch-name", default="")
|
|
parser.add_argument("--apply", action="store_true", help="create processed hardlink tree")
|
|
parser.add_argument("--write-file-manifest", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
source_root = args.source.resolve()
|
|
batch_name = args.batch_name or source_root.name
|
|
processed_batch_root = (args.processed_root / batch_name).resolve()
|
|
result_dir = (args.results_root / batch_name).resolve()
|
|
result_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
started_at = datetime.now().isoformat(timespec="seconds")
|
|
study_folders = sorted([p for p in source_root.iterdir() if p.is_dir()])
|
|
study_rows: list[StudyRow] = []
|
|
file_rows_all: list[dict[str, str]] = []
|
|
|
|
for index, folder in enumerate(study_folders, start=1):
|
|
print(f"[{index}/{len(study_folders)}] scan {folder.name}", flush=True)
|
|
study_row, file_rows = scan_study_folder(batch_name, folder, processed_batch_root)
|
|
study_rows.append(study_row)
|
|
if args.write_file_manifest:
|
|
file_rows_all.extend(file_rows)
|
|
|
|
ct_counts = Counter(row.ct_number for row in study_rows if row.ct_number)
|
|
target_counts = Counter(row.target_folder_name for row in study_rows if row.target_folder_name)
|
|
duplicate_ct_numbers = sorted([ct for ct, count in ct_counts.items() if count > 1])
|
|
duplicate_target_folders = sorted([name for name, count in target_counts.items() if count > 1])
|
|
|
|
for row in study_rows:
|
|
notes = [row.notes] if row.notes else []
|
|
if row.ct_number in duplicate_ct_numbers:
|
|
row.status = "error"
|
|
notes.append("duplicate_ct_number")
|
|
if row.target_folder_name in duplicate_target_folders:
|
|
row.status = "error"
|
|
notes.append("duplicate_target_folder")
|
|
row.notes = ";".join([note for note in notes if note])
|
|
|
|
study_dicts = [asdict(row) for row in study_rows]
|
|
study_fieldnames = list(asdict(study_rows[0]).keys()) if study_rows else list(StudyRow.__dataclass_fields__)
|
|
|
|
write_csv(result_dir / "study_manifest.csv", study_dicts, study_fieldnames)
|
|
write_csv(
|
|
result_dir / "rename_plan.csv",
|
|
[
|
|
{
|
|
"source_folder_name": row.source_folder_name,
|
|
"source_ct_number": row.source_ct_number,
|
|
"ct_number": row.ct_number,
|
|
"target_folder_name": row.target_folder_name,
|
|
"needs_ct_number_fix": row.needs_ct_number_fix,
|
|
"status": row.status,
|
|
"notes": row.notes,
|
|
}
|
|
for row in study_rows
|
|
],
|
|
[
|
|
"source_folder_name",
|
|
"source_ct_number",
|
|
"ct_number",
|
|
"target_folder_name",
|
|
"needs_ct_number_fix",
|
|
"status",
|
|
"notes",
|
|
],
|
|
)
|
|
if args.write_file_manifest:
|
|
write_csv(
|
|
result_dir / "file_manifest.csv",
|
|
file_rows_all,
|
|
[
|
|
"ct_number",
|
|
"source_folder_name",
|
|
"source_relative_path",
|
|
"processed_relative_path",
|
|
"sop_instance_uid",
|
|
"series_instance_uid",
|
|
"study_instance_uid",
|
|
"bytes",
|
|
],
|
|
)
|
|
|
|
ok_rows = [row for row in study_rows if row.status == "ok"]
|
|
error_rows = [row for row in study_rows if row.status != "ok"]
|
|
summary = {
|
|
"batch_name": batch_name,
|
|
"source_root": str(source_root),
|
|
"processed_batch_root": str(processed_batch_root),
|
|
"result_dir": str(result_dir),
|
|
"started_at": started_at,
|
|
"finished_scan_at": datetime.now().isoformat(timespec="seconds"),
|
|
"apply": args.apply,
|
|
"source_study_folder_count": len(study_folders),
|
|
"ok_study_count": len(ok_rows),
|
|
"error_study_count": len(error_rows),
|
|
"needs_ct_number_fix_count": sum(1 for row in study_rows if row.needs_ct_number_fix),
|
|
"duplicate_ct_numbers": duplicate_ct_numbers,
|
|
"duplicate_target_folders": duplicate_target_folders,
|
|
"dicom_file_count": sum(row.dicom_file_count for row in study_rows),
|
|
"total_file_count": sum(row.total_file_count for row in study_rows),
|
|
"total_bytes": sum(row.total_bytes for row in study_rows),
|
|
"linked_file_count": 0,
|
|
"already_linked_file_count": 0,
|
|
}
|
|
|
|
if error_rows:
|
|
summary_path = result_dir / "summary.json"
|
|
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f"ERROR: found {len(error_rows)} invalid study rows; see {summary_path}", file=sys.stderr)
|
|
return 2
|
|
|
|
if args.apply:
|
|
processed_batch_root.mkdir(parents=True, exist_ok=True)
|
|
linked_total = 0
|
|
already_total = 0
|
|
for index, row in enumerate(ok_rows, start=1):
|
|
print(f"[{index}/{len(ok_rows)}] link {row.source_folder_name} -> {row.target_folder_name}", flush=True)
|
|
linked, already_ok = hardlink_tree(Path(row.source_path), Path(row.processed_path))
|
|
linked_total += linked
|
|
already_total += already_ok
|
|
summary["linked_file_count"] = linked_total
|
|
summary["already_linked_file_count"] = already_total
|
|
summary["finished_apply_at"] = datetime.now().isoformat(timespec="seconds")
|
|
|
|
summary_path = result_dir / "summary.json"
|
|
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|