# Module to export deformation vector field to REG DICOM file
#
# APA, 4/3/2023
import SimpleITK as sitk
import os
from datetime import datetime
import numpy as np
from pydicom import dcmread
from pydicom.uid import generate_uid
from pydicom.dataset import Dataset, FileDataset, FileMetaDataset
from pydicom.sequence import Sequence
from pydicom.uid import ImplicitVRLittleEndian
import scipy.io as sio
import argparse
from random import randint
from cerr.dcm_export import iod_helper
org_root = '1.3.6.1.4.1.9590.100.1.2.'
[docs]
def getDcmTagVals(baseScanNum, planC, seriesOpts = {}):
pat_tags = {"PatientName": planC.scan[baseScanNum].scanInfo[0].patientName,
"PatientID": planC.scan[baseScanNum].scanInfo[0].patientID,
"PatientBirthDate": planC.scan[baseScanNum].scanInfo[0].patientBirthDate,
"PatientSex": planC.scan[baseScanNum].scanInfo[0].patientSex,
"PatientAge": "", #planC.scan[assocScanNum].scanInfo[0].patientAge
"PatientSize": planC.scan[baseScanNum].scanInfo[0].patientSize,
"PatientWeight": planC.scan[baseScanNum].scanInfo[0].patientWeight
}
study_tags = {"StudyDate": planC.scan[baseScanNum].scanInfo[0].studyDate,
"StudyTime": planC.scan[baseScanNum].scanInfo[0].studyTime,
"StudyDescription": planC.scan[baseScanNum].scanInfo[0].studyDescription,
"StudyInstanceUID": planC.scan[baseScanNum].scanInfo[0].studyInstanceUID,
"StudyID": "" #planC.scan[assocScanNum].scanInfo[0].studyID
}
if "seriesDescription" in seriesOpts:
seriesDescription = seriesOpts['seriesDescription']
else:
seriesDescription = "Generated by pyCERR"
dt = datetime.now()
if "SeriesDate" in seriesOpts:
SeriesDate = seriesOpts['SeriesDate']
else:
SeriesDate = dt.strftime("%Y%m%d")
if "SeriesTime" in seriesOpts:
SeriesTime = seriesOpts['SeriesTime']
else:
SeriesTime = dt.strftime("%H%M%S.%f")
series_tags = {
'Modality': 'REG',
'SeriesDate': SeriesDate,
'SeriesTime': SeriesTime,
'SeriesDescription': seriesDescription,
'SeriesInstanceUID': generate_uid(prefix=org_root),
'SeriesNumber': str(randint(9010, 9900))
}
equiqmt_tags = {"Manufacturer": "",
"ManufacturerModelName": "",
"InstitutionName": ""
}
content_tags = {"ContentDescription": "",
"ContentLabel": ""}
if "StructureSetLabel" in seriesOpts:
StructureSetLabel = seriesOpts['StructureSetLabel']
else:
StructureSetLabel = "RTStruct"
struct_set_tags = {"InstanceNumber": "",
"StructureSetDescription": "",
"StructureSetLabel": StructureSetLabel
}
return pat_tags, study_tags, series_tags, equiqmt_tags, content_tags, struct_set_tags
[docs]
def createREGDataset(base_series_data, reg_out_dcm_file) -> FileDataset:
ds = generateBaseDataset(reg_out_dcm_file)
addStudyAndSeriesInformation(ds, base_series_data)
addPatientInformation(ds, base_series_data)
addRefSeriesSequence(ds, base_series_data)
addContentInformation(ds)
return ds
[docs]
def generateBaseDataset(file_name) -> FileDataset:
file_meta = getFileMeta()
ds = FileDataset(file_name, {}, file_meta=file_meta, preamble=b"\0" * 128)
addRequiredElementsToDs(ds)
addSequenceListsToDs(ds)
return ds
[docs]
def addRequiredElementsToDs(ds: FileDataset):
dt = datetime.now()
# Append data elements required by the DICOM standarad
ds.SpecificCharacterSet = "ISO_IR 192" # "ISO_IR 100"
ds.InstanceCreationDate = dt.strftime("%Y%m%d")
ds.InstanceCreationTime = dt.strftime("%H%M%S.%f")
ds.Modality = "REG"
ds.Manufacturer = "reg2dcm"
ds.ManufacturerModelName = "reg2dcm"
ds.InstitutionName = "MSKCC"
# Set the transfer syntax
ds.is_little_endian = True
ds.is_implicit_VR = True
# Set values already defined in the file meta
ds.SOPClassUID = ds.file_meta.MediaStorageSOPClassUID
ds.SOPInstanceUID = ds.file_meta.MediaStorageSOPInstanceUID
[docs]
def addSequenceListsToDs(ds: FileDataset):
ds.DeformableRegistrationSequence = Sequence()
[docs]
def addContentInformation(ds: FileDataset):
dt = datetime.now()
ds.ContentCreatorName = ''
ds.ContentDate = dt.strftime("%Y%m%d")
ds.ContentTime = dt.strftime("%H%M%S.%f")
ds.ContentDescription = 'AI REGISTRATION'
ds.ContentLabel = "REGISTRATION"
[docs]
def getRefImgSeq(series_data):
refd_img_seq = Sequence()
for img in series_data:
ref_img_ds = Dataset()
ref_img_ds.ReferencedSOPClassUID = getattr(img, 'SOPClassUID', generate_uid())
ref_img_ds.ReferencedSOPInstanceUID = getattr(img, 'SOPInstanceUID', generate_uid())
refd_img_seq.append(ref_img_ds)
return refd_img_seq
[docs]
def addRefSeriesSequence(ds: FileDataset, series_data):
ref_series_seq = Sequence()
ref_series_ds = Dataset()
ref_series_ds.ReferencedInstanceSequence = getRefImgSeq(series_data)
ref_series_ds.SeriesInstanceUID = getattr(series_data[0], 'SeriesInstanceUID', generate_uid())
ref_series_seq.append(ref_series_ds)
ds.ReferencedSeriesSequence = ref_series_seq
return
# return FOR # Frame of reference UID
[docs]
def getREGTypeCodeSeq(series_type):
reg_type_seq = Sequence()
reg_type_ds = Dataset()
if series_type == "base":
reg_type_ds.CodingSchemeDesignator = "DCM"
reg_type_ds.CodeValue = "125021"
reg_type_ds.CodeMeaning = "Frame of Reference Identity"
elif series_type == "mov":
reg_type_ds.CodingSchemeDesignator = "DCM"
reg_type_ds.CodeValue = "125024"
reg_type_ds.CodeMeaning = "Image Content-based Alignment"
reg_type_seq.append(reg_type_ds)
return reg_type_seq
[docs]
def getDVFGridSeq(dvf_nii_file):
dvf_grid_seq = Sequence()
grid_ds = Dataset()
dvf_matrix, origin, orient, dim, res, original_orient_str = getDVFMetadata(dvf_nii_file)
grid_ds.ImagePositionPatient = list(origin)
grid_ds.ImageOrientationPatient = orient
grid_ds.GridDimensions = dim
grid_ds.GridResolution = res
dx = dvf_matrix[:, :, :, 0]
dy = dvf_matrix[:, :, :, 1]
dz = dvf_matrix[:, :, :, 2]
# plot dvf for QA
# dx_slice = dx[:, :, 50]
# import matplotlib.pyplot as plt
# plt.imshow(dx_slice)
dx = dx.ravel(order="C")
dy = dy.ravel(order="C")
dz = dz.ravel(order="C")
num_elements = np.prod(dim) * 3
vector_grid_matrix = np.array([dx, dy, dz], dtype=np.single).transpose().reshape(num_elements, 1)
grid_ds.VectorGridData = vector_grid_matrix
dvf_grid_seq.append(grid_ds)
# # Save DVF to nii file - can be used to expand the module to convert dicom REG to nii
# import nibabel as nib
# dvf_img = nib.load(dvf_nii_file)
# dvf_data = dvf_img.get_fdata()
# affine = dvf_img.affine
# # Create 5cm deformation in L-R direction
# dvf_data[:256,:,:,0,0] = 50.0
# dvf_data[257:,:,:,0,0] = 0.0
# dvf_data[:,:,:,0,1] = 0.0
# dvf_data[:,:,:,0,2] = 0.0
# dvf_updated_img = nib.Nifti1Image(dvf_data, affine=affine)
# dvf_updated_img = nib.Nifti1Image(dvf_data, dvf_img.affine, dvf_img.header)
# save_dir = os.path.dirname(dvf_nii_file)
# save_filename = os.path.join(save_dir,'update_LR_5cm.nii')
# nib.save(dvf_updated_img, save_filename)
return dvf_grid_seq
[docs]
def getMatrixSeq(deform_matrix_file):
dvf_matrix_seq = Sequence()
dvf_matrix_ds = Dataset()
transform_matrix = loadMatrixFile(deform_matrix_file)
dvf_matrix_ds.FrameOfReferenceTransformationMatrixType = "RIGID"
dvf_matrix_ds.FrameOfReferenceTransformationMatrix = transform_matrix
dvf_matrix_seq.append(dvf_matrix_ds)
return dvf_matrix_seq
[docs]
def loadMatrixFile(deform_matrix_file):
if deform_matrix_file is not None and os.path.exists(deform_matrix_file):
tf_contents = sio.loadmat(deform_matrix_file)
affine_3x3 = tf_contents["AffineTransform_double_3_3"]
center = tf_contents["fixed"]
matrix = np.reshape(affine_3x3[:9],(3,3))
translation = affine_3x3[9:12]
last_row = np.expand_dims(np.array([0,0,0,1]), axis=1).T
offset = translation + center - np.dot(matrix,center)
matrix = np.concatenate((matrix,offset),axis=1)
matrix = np.concatenate((matrix,last_row),axis=0)
transform_matrix = list(matrix.flatten())
else:
transform_matrix = np.eye(4)
transform_matrix = list(transform_matrix.reshape((16, 1)))
return transform_matrix
[docs]
def loadSortedSeriesData(dicom_series_path):
series_data = []
for root, _, files in os.walk(dicom_series_path):
for file in files:
try:
ds = dcmread(os.path.join(root, file))
if hasattr(ds, "pixel_array"):
series_data.append(ds)
except Exception:
# Not a valid DICOM file
continue
series_data.sort(key=getSlicePosition, reverse=False)
return series_data
[docs]
def getSlicePosition(series_slice: Dataset):
_, _, slice_direction = get_slice_directions(series_slice)
return np.dot(slice_direction, series_slice.ImagePositionPatient)
[docs]
def get_slice_directions(series_slice: Dataset):
orientation = series_slice.ImageOrientationPatient
row_direction = np.array(orientation[:3])
column_direction = np.array(orientation[3:])
slice_direction = np.cross(row_direction, column_direction)
if not np.allclose(
np.dot(row_direction, column_direction), 0.0, atol=1e-3
) or not np.allclose(np.linalg.norm(slice_direction), 1.0, atol=1e-3):
raise Exception("Invalid Image Orientation (Patient) attribute")
return row_direction, column_direction, slice_direction
[docs]
def create(filePath, baseScanNum, movScanNum, dvf_nii_file, reg_out_dcm_file, pre_matrix_file,
post_matrix_file, planC, seriesOpts = {}):
# Get related UIDs for structNumV
pat_tags, study_tags, series_tags, equiqmt_tags, content_tags, struct_set_tags = \
getDcmTagVals(baseScanNum, planC, seriesOpts)
# Initialize RTSTRUCT series
# ds = create_reg_dataset(base_series_data, filePath)
file_meta = iod_helper.getFileMeta('REG')
ds = FileDataset(filePath, {}, file_meta=file_meta, preamble=b"\0" * 128)
ds = iod_helper.addSOPCommonTags(ds)
# Add Patient tags
ds = iod_helper.addPatientTags(ds, pat_tags)
# Add Study tags
ds = iod_helper.addStudyTags(ds, study_tags)
# Add Series tags
ds = iod_helper.addSeriesTags(ds, series_tags)
# Add Equipment tags
ds = iod_helper.addEquipmentTags(ds, equiqmt_tags)
# Add Content tags
ds = iod_helper.addContentTags(ds, content_tags)
# Add Structure Set tags
ds = iod_helper.addStructureSetTags(ds, struct_set_tags)
print("Writing REG file ...", filePath)
ds.save_as(filePath)
print("File saved.")
[docs]
def createDVF(base_dir, mov_dir, dvf_nii_file, pre_matrix_file,
post_matrix_file, reg_out_dcm_file):
# base_dir = r'H:\Public\huj\ForAditya\DIR\Case1\2018-05'
# mov_dir = r'H:\Public\huj\ForAditya\DIR\Case1\2018-04'
# dvf_nii_file = r'H:\Public\huj\ForAditya\DIR\Case1\Plastimatch\vf.mha'
# reg_out_dcm_file = r'H:\Public\huj\ForAditya\DIR\Case1\test_reg_export\plastimatch_dvf.dcm'
# pre_matrix_file = ""
# post_matrix_file = ""
# base_dir = r'L:\Aditya\forEve\registration_HN_sample_data\pet_ct_planningCT_examples\10031975\petct_10031975\CT_20130509_CT_HN'
# mov_dir = r'L:\Aditya\forEve\registration_HN_sample_data\pet_ct_planningCT_examples\10031975\planningCT_10031975\BOT_107_\CT_20130509_UNKNOWN'
# dvf_nii_file = r'L:\Aditya\forEve\registration_HN_sample_data\pet_ct_planningCT_examples\10031975\registration\CT_to_PET_1Warp.nii.gz'
# pre_matrix_file = r'L:\Aditya\forEve\registration_HN_sample_data\pet_ct_planningCT_examples\10031975\registration\CT_to_PET_0GenericAffine.mat'
# post_matrix_file = ''
# reg_out_dcm_file = r'L:\Aditya\forEve\registration_HN_sample_data\pet_ct_planningCT_examples\10031975\dicom_export\ants\ANTS_deform.dcm'
# base_dir = r"L:\Aditya\forEve\registration_HN_sample_data\0617_examples\0617-259694\09-09-2000-79111\1-CTs from rtog conversion-92395"
# mov_dir = r"L:\Aditya\forEve\registration_HN_sample_data\0617_examples\0617-292370\09-09-2000-35932\1-CTs from rtog conversion-33570"
# dvf_nii_file = r"L:\Aditya\forEve\registration_HN_sample_data\RTOG0617_reg\f259694_m292370_1Warp.nii.gz"
# pre_matrix_file = r"L:\Aditya\forEve\registration_HN_sample_data\RTOG0617_reg\f259694_m292370_0GenericAffine.mat"
# post_matrix_file = ""
# reg_out_dcm_file = r"L:\Aditya\forEve\registration_HN_sample_data\RTOG0617_reg\dicom_dvf_out\f259694_m292370_dvf.dcm"
# base_dir = r"L:\Aditya\forEve\registration_HN_sample_data\0617_examples\0617-292370\09-09-2000-35932\1-CTs from rtog conversion-33570"
# mov_dir = r"L:\Aditya\forEve\registration_HN_sample_data\0617_examples\0617-305105\09-09-2000-66193\1-CTs from rtog conversion-36569"
# dvf_nii_file = r"L:\Aditya\forEve\registration_HN_sample_data\RTOG0617_reg\f292370_m305105_1Warp.nii.gz"
# pre_matrix_file = r"L:\Aditya\forEve\registration_HN_sample_data\RTOG0617_reg\f292370_m305105_0GenericAffine.mat"
# post_matrix_file = ""
# reg_out_dcm_file = r"L:\Aditya\forEve\registration_HN_sample_data\RTOG0617_reg\dicom_dvf_out\f292370_m305105_dvf.dcm"
# Jue's dataset
# mov_dir = r'M:\Aditya\forNeelam\registration_sample_data\baseline'
# base_dir = r'M:\Aditya\forNeelam\registration_sample_data\timept1'
# dvf_nii_file = r'M:\Aditya\forNeelam\registration_sample_data\cerrFile^38167419_202305093535_MR_DVF_first_scan_3D.nii.gz'
# #dvf_nii_file = r'L:\Aditya\forEve\registration_HN_sample_data\pet_ct_planningCT_examples\10031975\registration\CT_to_PET_1Warp.nii.gz'
# dvf_nii_file = r"L:\Aditya\forEve\registration_HN_sample_data\RTOG0617_reg\f292370_m305105_1Warp.nii.gz"
# dvf_nii_file = r"\\vpensmph\deasylab1\Aditi\forAditya\testNiiExport\test_dvf_nii_apa_pt1\cerrFile^38167419_202305093535_MR DVF_first_scan_3D.nii.gz"
# pre_matrix_file = ""
# post_matrix_file = ""
# reg_out_dcm_file = r'\\vpensmph\deasylab2\Aditya\forNeelam\registration_sample_data\REG_output\proRSeg_REG.dcm'
# dcm_dir = r"\\vpensmph\deasylab1\Aditi\forJue\MRI_Pancreas_Fullshot_AnatomicCtxShape_v2\compare_out\dicom_inputs\38167419"
# mov_dir = os.path.join(dcm_dir,"38167419_1_3_46_670589_11_79106_5_0_1204_2022040414131734337")
# base_dir = os.path.join(dcm_dir,"38167419_1_3_46_670589_11_79106_5_0_1204_2022040512011894235")
# pre_matrix_file = ""
# post_matrix_file = ""
# dvf_nii_file = r"\\vpensmph\deasylab2\Aditya\forNeelam\registration_sample_data\REG_output\cerrFile^38167419_MR_DVF_first_scan_3D.nii.gz"
# reg_out_dcm_file = r'\\vpensmph\deasylab2\Aditya\forNeelam\registration_sample_data\REG_output\proRSeg_REG.dcm'
# Mermaid dvf
base_dir = r"\\vpensmph\deasylab1\Aditya\forEve\dvf_mri\test_pt1\35558483\CBCT01_OG\CT_OnlineMatchResliced"
mov_dir = r"\\vpensmph\deasylab1\Aditya\forEve\dvf_mri\test_pt1\35558483\planCT_OG\CT_1900"
pre_matrix_file = ""
post_matrix_file = ""
dvf_nii_file = r"\\vpensmph\deasylab1\Aditya\forEve\dvf_mri\test_pt1\phi.nii"
reg_out_dcm_file = r'\\vpensmph\deasylab1\Aditya\forEve\dvf_mri\test_pt1\reg_out\mermaidREG.dcm'
base_series_data = loadSortedSeriesData(base_dir)
mov_series_data = loadSortedSeriesData(mov_dir)
ds = createREGDataset(base_series_data, reg_out_dcm_file)
ds.FrameOfReferenceUID = getattr(base_series_data[0], "FrameOfReferenceUID", "")
deformable_reg_seq = createDeformSeq(base_series_data, mov_series_data, dvf_nii_file, pre_matrix_file, post_matrix_file)
ds.DeformableRegistrationSequence = deformable_reg_seq
print("Writing REG file", reg_out_dcm_file)
ds.save_as(reg_out_dcm_file)
print("File saved.")
if __name__ == '__main__':
# Parse input arguments
parser = argparse.ArgumentParser()
parser.add_argument('-b','--base_dir', required=True, help='specify the path to directory containing reference images')
parser.add_argument('-m','--mov_dir', required=True, help='specify the path to directory containing moving images')
parser.add_argument('-d','--dvf_nii_file', required=True, help='specify the path to file (ITK readable such as nii, mha, mhd) '
'containing Deformation Vector Field. The deformation vector field '
'should have 3-components for the 3D dataset representing deformations '
'in x, y and z-directions respectively in DICOM LPS coordinate system.')
parser.add_argument('-pr','--pre_matrix_file', required=False, help='specify the path to ITK compatible txt file'
' containing pre-deformation rigid registration')
parser.add_argument('-po','--post_matrix_file', required=False, help='specify the path to ITK compatible txt file'
' containing post-deformation rigid registration')
parser.add_argument('-o','--reg_out_dcm_file', required=True, help='specify the path to output DICOM file'
' containing REG modality')
opt = parser.parse_args()
createDVF(opt.base_dir, opt.mov_dir, opt.dvf_nii_file, opt.pre_matrix_file,
opt.post_matrix_file, opt.reg_out_dcm_file)