File size: 6,809 Bytes
377b913
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
"""Minimal, dependency-free DICOM I/O (explicit VR little-endian).

This is NOT a full DICOM implementation. It handles exactly the tags needed to
reconstruct a 3D MR volume: geometry, ordering, and 16-bit pixel data. It exists
so (a) the pipeline can be tested without pydicom/SimpleITK, and (b) it can serve
as a last-resort fallback backend in prepare_data.py. In production you should
prefer SimpleITK or pydicom (auto-detected there).
"""
import struct
import numpy as np

# tags we care about  (group, element) -> name
TAGS = {
    (0x0008, 0x0060): "Modality",
    (0x0020, 0x000D): "StudyInstanceUID",
    (0x0020, 0x000E): "SeriesInstanceUID",
    (0x0020, 0x0013): "InstanceNumber",
    (0x0020, 0x0032): "ImagePositionPatient",
    (0x0020, 0x0037): "ImageOrientationPatient",
    (0x0028, 0x0010): "Rows",
    (0x0028, 0x0011): "Columns",
    (0x0028, 0x0030): "PixelSpacing",
    (0x0018, 0x0050): "SliceThickness",
    (0x0028, 0x0100): "BitsAllocated",
    (0x0028, 0x0101): "BitsStored",
    (0x0028, 0x0103): "PixelRepresentation",
    (0x7FE0, 0x0010): "PixelData",
}
_EXPLICIT_LONG_VR = {"OB", "OW", "OF", "SQ", "UT", "UN"}


def is_dicom(path):
    """True if the file has the DICM magic at byte 128 (works for extension-less files)."""
    try:
        with open(path, "rb") as f:
            f.seek(128)
            return f.read(4) == b"DICM"
    except Exception:
        return False


def read_dicom(path):
    """Parse a single explicit-VR-LE DICOM file into a dict of the TAGS above."""
    with open(path, "rb") as f:
        buf = f.read()
    if buf[128:132] != b"DICM":
        raise ValueError("not a preambled DICOM: " + path)
    i = 132
    out = {}
    n = len(buf)
    while i + 8 <= n:
        group, elem = struct.unpack_from("<HH", buf, i); i += 4
        vr = buf[i:i + 2].decode("ascii", "ignore"); i += 2
        if vr in _EXPLICIT_LONG_VR:
            i += 2  # reserved
            (length,) = struct.unpack_from("<I", buf, i); i += 4
        elif vr.isalpha() and vr.isupper() and len(vr) == 2:
            (length,) = struct.unpack_from("<H", buf, i); i += 2
        else:
            # implicit VR fallback: the 2 bytes we read as VR are really length low half
            i -= 2
            (length,) = struct.unpack_from("<I", buf, i); i += 4
            vr = ""
        if length == 0xFFFFFFFF:  # undefined length (SQ) - skip, we don't need it
            length = 0
        val = buf[i:i + length]; i += length
        name = TAGS.get((group, elem))
        if name is None:
            continue
        if name == "PixelData":
            out["PixelData"] = val
        elif name in ("Rows", "Columns", "BitsAllocated", "BitsStored", "PixelRepresentation"):
            out[name] = struct.unpack_from("<H", val, 0)[0] if len(val) >= 2 else 0
        else:
            out[name] = val.decode("ascii", "ignore").strip("\x00 ")
    return out


def _ds_floats(s, k):
    try:
        return [float(x) for x in s.split("\\")][:k]
    except Exception:
        return [0.0] * k


def volume_from_series(paths):
    """Stack a list of single-slice DICOM paths into an ordered 3D array + spacing.

    Ordering: by projection onto the slice normal (from ImageOrientationPatient +
    ImagePositionPatient) when available, else by InstanceNumber, else filename.
    Returns (volume[X,Y,Z] float32, spacing=(sx,sy,sz)).
    """
    slices = []
    for p in paths:
        try:
            d = read_dicom(p)
            if "PixelData" in d and "Rows" in d:
                slices.append((p, d))
        except Exception:
            continue
    if not slices:
        raise ValueError("no readable slices")

    def sort_key(item):
        _, d = item
        ipp = _ds_floats(d.get("ImagePositionPatient", ""), 3)
        iop = _ds_floats(d.get("ImageOrientationPatient", ""), 6)
        if len(ipp) == 3 and len(iop) == 6:
            r = np.array(iop[:3]); c = np.array(iop[3:]); nrm = np.cross(r, c)
            return float(np.dot(nrm, ipp))
        try:
            return float(d.get("InstanceNumber", 0))
        except Exception:
            return 0.0

    slices.sort(key=sort_key)
    rows = slices[0][1]["Rows"]; cols = slices[0][1]["Columns"]
    signed = slices[0][1].get("PixelRepresentation", 0) == 1
    dtype = np.int16 if signed else np.uint16
    planes = []
    for _, d in slices:
        arr = np.frombuffer(d["PixelData"], dtype=dtype)[: rows * cols].reshape(rows, cols)
        planes.append(arr.astype(np.float32))
    vol = np.stack(planes, axis=-1)                 # (rows, cols, nslices) = (Y, X, Z)
    vol = np.transpose(vol, (1, 0, 2))              # -> (X, Y, Z)
    ps = _ds_floats(slices[0][1].get("PixelSpacing", "1\\1"), 2) or [1.0, 1.0]
    # slice spacing from position delta if possible
    if len(slices) > 1:
        z0 = sort_key(slices[0]); z1 = sort_key(slices[-1])
        sz = abs(z1 - z0) / max(1, len(slices) - 1) or _ds_floats(slices[0][1].get("SliceThickness", "1"), 1)[0]
    else:
        sz = _ds_floats(slices[0][1].get("SliceThickness", "1"), 1)[0] or 1.0
    spacing = (float(ps[1]), float(ps[0]), float(sz))
    return vol, spacing


# --------------------------------------------------------------------------
# Minimal writer (used only by the test harness to synthesize DICOM series)
# --------------------------------------------------------------------------
def _elem(group, elem, vr, val):
    if isinstance(val, str):
        val = val.encode("ascii")
        if len(val) % 2:
            val += b" " if vr not in ("UI",) else b"\x00"
    head = struct.pack("<HH", group, elem) + vr.encode("ascii")
    if vr in _EXPLICIT_LONG_VR:
        head += b"\x00\x00" + struct.pack("<I", len(val))
    else:
        head += struct.pack("<H", len(val))
    return head + val


def write_dicom(path, arr2d, series_uid, instance, z, spacing=(1, 1, 1)):
    arr2d = arr2d.astype(np.uint16)
    rows, cols = arr2d.shape
    body = b"".join([
        _elem(0x0008, 0x0060, "CS", "MR"),
        _elem(0x0020, 0x000D, "UI", "1.2.3.4.5"),
        _elem(0x0020, 0x000E, "UI", series_uid),
        _elem(0x0020, 0x0013, "IS", str(instance)),
        _elem(0x0020, 0x0032, "DS", f"0\\0\\{z}"),
        _elem(0x0020, 0x0037, "DS", "1\\0\\0\\0\\1\\0"),
        _elem(0x0028, 0x0010, "US", struct.pack("<H", rows)),
        _elem(0x0028, 0x0011, "US", struct.pack("<H", cols)),
        _elem(0x0028, 0x0030, "DS", f"{spacing[1]}\\{spacing[0]}"),
        _elem(0x0018, 0x0050, "DS", str(spacing[2])),
        _elem(0x0028, 0x0100, "US", struct.pack("<H", 16)),
        _elem(0x0028, 0x0101, "US", struct.pack("<H", 16)),
        _elem(0x0028, 0x0103, "US", struct.pack("<H", 0)),
        _elem(0x7FE0, 0x0010, "OW", arr2d.tobytes()),
    ])
    with open(path, "wb") as f:
        f.write(b"\x00" * 128 + b"DICM" + body)