core/dicom.py

# 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