claimflow-api / app /services /dicom_preview.py
Minifigures's picture
fix: seed skips in-place DICOM rewrite (app/services/dicom_preview.py)
8463fc5 verified
Raw
History Blame Contribute Delete
4.29 kB
"""DICOM intake: PHI de-identification at rest, safe metadata, PNG preview.
Uploaded studies are rewritten in place with every denylisted PHI tag blanked
before anything else reads them; only a small allowlisted metadata dict (with
the study date truncated to a year) is ever persisted alongside the document.
"""
from pathlib import Path
import pydicom
from PIL import Image
from pydicom.datadict import tag_for_keyword
PHI_DENYLIST: tuple[str, ...] = (
"PatientName",
"PatientID",
"PatientBirthDate",
"OtherPatientIDs",
"OtherPatientNames",
"PatientAddress",
"PatientTelephoneNumbers",
"InstitutionName",
"InstitutionAddress",
"ReferringPhysicianName",
"PerformingPhysicianName",
"OperatorsName",
"AccessionNumber",
)
_SAFE_STR_FIELDS = ("Modality", "Manufacturer", "ManufacturerModelName", "SoftwareVersions")
_SAFE_INT_FIELDS = ("Rows", "Columns")
_MAGIC_PNG = b"\x89PNG\r\n\x1a\n"
_MAGIC_JPEG = b"\xff\xd8\xff"
_MAGIC_PDF = b"%PDF-"
_MAGIC_DICM = b"DICM"
def sniff_kind(path: Path, mime: str) -> str:
"""Classify a stored upload by magic bytes.
The client extension and mime (passed for logging/diagnostics only) are never
trusted alone. Returns 'dicom' | 'png' | 'jpeg' | 'pdf'.
"""
with path.open("rb") as fh:
head = fh.read(132)
if len(head) >= 132 and head[128:132] == _MAGIC_DICM:
return "dicom"
if head.startswith(_MAGIC_PNG):
return "png"
if head.startswith(_MAGIC_JPEG):
return "jpeg"
if head.startswith(_MAGIC_PDF):
return "pdf"
raise ValueError("unsupported file type")
def process_dicom(path: Path, *, rewrite: bool = True) -> tuple[dict, str | None]:
"""De-identify a DICOM file *in place*, extract safe metadata, render a preview.
Every denylisted PHI tag present is blanked and the file rewritten on disk
before returning, so the study is de-identified at rest. Returns
(meta, preview_path); preview_path is None when pixel data is missing or
unreadable. Raises ValueError for non-DICOM input.
``rewrite=False`` skips the at-rest rewrite for callers whose input is
synthetic and PHI-free by construction (the demo seeder); metadata and
preview extraction behave identically.
"""
try:
ds = pydicom.dcmread(path, force=True)
except Exception as exc:
raise ValueError("not a valid DICOM file") from exc
if len(ds) == 0 or "TransferSyntaxUID" not in ds.file_meta:
raise ValueError("not a valid DICOM file")
for keyword in PHI_DENYLIST:
tag = tag_for_keyword(keyword)
if tag is not None and tag in ds:
ds[tag].value = ""
meta: dict[str, object] = {}
for keyword in _SAFE_STR_FIELDS:
value = ds.get(keyword)
if value is not None and str(value):
meta[keyword] = str(value)
for keyword in _SAFE_INT_FIELDS:
value = ds.get(keyword)
if value is not None:
meta[keyword] = int(value)
study_date = str(ds.get("StudyDate") or "")
if study_date:
meta["StudyDate"] = study_date[:4] # year only — never persist the full date
pixel_spacing = ds.get("PixelSpacing")
if pixel_spacing is not None:
meta["PixelSpacing"] = str(pixel_spacing)
if rewrite:
ds.save_as(path, enforce_file_format=True) # rewrite: de-identified at rest
with path.open("rb") as fh:
head = fh.read(132)
if len(head) < 132 or head[128:132] != _MAGIC_DICM:
import logging
logging.getLogger("claimflow.dicom").error(
"post-rewrite verification failed for %s: size=%d head=%s",
path,
path.stat().st_size,
head[:16].hex(),
)
preview = path.with_suffix(".png")
try:
arr = ds.pixel_array.astype("float32")
except Exception:
return meta, None
lo, hi = float(arr.min()), float(arr.max())
arr = (arr - lo) / (hi - lo) * 255.0 if hi > lo else arr * 0.0
arr8 = arr.astype("uint8")
if arr8.ndim == 3: # color or multi-frame: preview the first plane
arr8 = arr8[..., 0] if arr8.shape[-1] in (3, 4) else arr8[0]
Image.fromarray(arr8).save(preview)
return meta, str(preview)