File size: 4,749 Bytes
9a75c73 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | # mediagent/core/dicom.py
"""
DICOM file parser for MediAgent.
Extracts pixel data + clinical metadata from .dcm files,
converts to base64 PNG for the vision pipeline, and returns
structured metadata to pre-populate the intake form.
"""
import base64
import io
import logging
from typing import Any, Dict, Optional, Tuple
logger = logging.getLogger(__name__)
def parse_dicom(file_bytes: bytes) -> Tuple[str, Dict[str, Any]]:
"""
Parse a DICOM (.dcm) file.
Returns:
(base64_image_string, metadata_dict)
base64_image_string: "data:image/png;base64,..." ready for vision pipeline
metadata_dict: extracted clinical metadata for intake pre-population
"""
try:
import pydicom
import numpy as np
from PIL import Image
except ImportError as e:
raise ImportError(f"DICOM support requires pydicom, numpy, Pillow: {e}")
ds = pydicom.dcmread(io.BytesIO(file_bytes), force=True)
# ── Metadata extraction ───────────────────────────────────────────────────
metadata: Dict[str, Any] = {}
_tag_map = {
"PatientName": "patient_name",
"PatientID": "patient_id",
"PatientBirthDate": "birth_date",
"PatientSex": "sex",
"PatientAge": "age_str",
"StudyDate": "study_date",
"StudyDescription": "study_description",
"SeriesDescription": "series_description",
"Modality": "modality",
"InstitutionName": "institution",
"Manufacturer": "manufacturer",
"ManufacturerModelName": "device_model",
"KVP": "kvp",
"ExposureTime": "exposure_time_ms",
"SliceThickness": "slice_thickness_mm",
"BodyPartExamined": "body_part",
"StudyInstanceUID": "study_uid",
"SOPInstanceUID": "instance_uid",
"Rows": "image_rows",
"Columns": "image_cols",
"PixelSpacing": "pixel_spacing_mm",
}
for dicom_tag, key in _tag_map.items():
try:
val = getattr(ds, dicom_tag, None)
if val is not None:
metadata[key] = str(val)
except Exception:
pass
# Normalise age: DICOM age strings look like "045Y", "006M", "010D"
age: Optional[int] = None
age_str = metadata.pop("age_str", None)
if age_str:
try:
if age_str.endswith("Y"):
age = int(age_str[:-1])
elif age_str.endswith("M"):
age = max(0, int(int(age_str[:-1]) / 12))
except ValueError:
pass
if age is not None:
metadata["age"] = age
# Normalise sex: DICOM uses M/F/O
sex = metadata.get("sex", "")
if sex and sex.upper() in ("M", "F", "O"):
metadata["sex"] = sex.upper()
else:
metadata.pop("sex", None)
# ── Pixel data → PNG base64 ───────────────────────────────────────────────
try:
pixel_array = ds.pixel_array.astype(float)
except Exception as e:
raise ValueError(f"Could not read DICOM pixel data: {e}")
# MONOCHROME1 means bright = low value → invert
photometric = str(getattr(ds, "PhotometricInterpretation", "MONOCHROME2")).strip()
if photometric == "MONOCHROME1":
pixel_array = pixel_array.max() - pixel_array
# Normalise to 0–255
p_min, p_max = pixel_array.min(), pixel_array.max()
if p_max > p_min:
pixel_array = ((pixel_array - p_min) / (p_max - p_min) * 255).astype("uint8")
else:
pixel_array = pixel_array.astype("uint8")
# Handle grayscale, RGB, multi-frame (take first frame)
if pixel_array.ndim == 3 and pixel_array.shape[0] > 3:
pixel_array = pixel_array[0] # first frame of multi-frame
if pixel_array.ndim == 2:
img = Image.fromarray(pixel_array, mode="L").convert("RGB")
else:
img = Image.fromarray(pixel_array.astype("uint8"))
buf = io.BytesIO()
img.save(buf, format="PNG", optimize=True)
b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
base64_image = f"data:image/png;base64,{b64}"
logger.info(
f"DICOM parsed | modality={metadata.get('modality','?')} "
f"body_part={metadata.get('body_part','?')} "
f"size={metadata.get('image_rows','?')}x{metadata.get('image_cols','?')}"
)
return base64_image, metadata
|