Source code for cerr.dcm_export.reg_iod

# 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 getFileMeta() -> FileMetaDataset: file_meta = FileMetaDataset() file_meta.FileMetaInformationGroupLength = 202 file_meta.FileMetaInformationVersion = b"\x00\x01" file_meta.TransferSyntaxUID = ImplicitVRLittleEndian file_meta.MediaStorageSOPClassUID = "1.2.840.10008.5.1.4.1.1.66.3" file_meta.MediaStorageSOPInstanceUID = ( generate_uid() ) # TODO find out random generation is fine file_meta.ImplementationClassUID = "1.2.246.352.70.2.1.160.3" return file_meta
[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 addStudyAndSeriesInformation(ds: FileDataset, series_data): dt = datetime.now() reference_ds = series_data[0] # All elements in series should have the same data ds.StudyDate = reference_ds.StudyDate ds.StudyTime = getattr(reference_ds, "StudyTime", "") ds.SeriesDate = dt.strftime("%Y%m%d") #getattr(reference_ds, "SeriesDate", "") ds.SeriesTime = dt.strftime("%H%M%S.%f") #getattr(reference_ds, "SeriesTime", "") ds.StudyDescription = getattr(reference_ds, "StudyDescription", "") ds.SeriesDescription = "AI REG" #getattr(reference_ds, "SeriesDescription", "") ds.StudyInstanceUID = reference_ds.StudyInstanceUID ds.SeriesInstanceUID = generate_uid() ds.StudyID = reference_ds.StudyID ds.SeriesNumber = str(randint(9010, 9900))
[docs] def addPatientInformation(ds: FileDataset, series_data): reference_ds = series_data[0] # All elements in series should have the same data ds.PatientName = getattr(reference_ds, "PatientName", "") ds.PatientID = getattr(reference_ds, "PatientID", "") ds.PatientBirthDate = getattr(reference_ds, "PatientBirthDate", "") ds.PatientSex = getattr(reference_ds, "PatientSex", "") ds.PatientAge = getattr(reference_ds, "PatientAge", "") ds.PatientSize = getattr(reference_ds, "PatientSize", "") ds.PatientWeight = getattr(reference_ds, "PatientWeight", "")
[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
[docs] def getDVFMetadata(dvf_nii_file): reader = sitk.ImageFileReader() reader.SetFileName(dvf_nii_file) reader.LoadPrivateTagsOn() reader.ReadImageInformation() image = reader.Execute() img_ori = image.GetDirection() original_orient_str = sitk.DICOMOrientImageFilter_GetOrientationFromDirectionCosines(img_ori) image = sitk.DICOMOrient(image,"LPS") # temp dvf_matrix = sitk.GetArrayFromImage(image) origin = list(image.GetOrigin()) orient = list(image.GetDirection())[:6] dim = list(image.GetSize()) res = list(image.GetSpacing()) return dvf_matrix, origin, orient, dim, res, original_orient_str
[docs] def getImgDirMetadata(base_dir): pass
# return FOR # Frame of reference UID
[docs] def createDeformSeq(base_series_data, mov_series_data, dvf_nii_file, pre_def_matrix_file=None, post_def_matrix_file=None): # return deformable_reg_seq def_seq = Sequence() mov_ds = Dataset() base_ds = Dataset() source_FOR_UID = getattr(base_series_data[0], "FrameOfReferenceUID", "") base_ds.SourceFrameOfReferenceUID = source_FOR_UID base_ds.RegistrationTypeCodeSequence = getREGTypeCodeSeq('base') base_ds.ReferencedImageSequence = getRefImgSeq(base_series_data) def_seq.append(base_ds) source_FOR_UID = getattr(mov_series_data[0], "FrameOfReferenceUID", "") mov_ds.SourceFrameOfReferenceUID = source_FOR_UID mov_ds.DeformableRegistrationGridSequence = getDVFGridSeq(dvf_nii_file) mov_ds.PreDeformationMatrixRegistrationSequence = getMatrixSeq(pre_def_matrix_file) mov_ds.PostDeformationMatrixRegistrationSequence = getMatrixSeq(post_def_matrix_file) mov_ds.RegistrationTypeCodeSequence = getREGTypeCodeSeq('mov') mov_ds.ReferencedImageSequence = getRefImgSeq(mov_series_data) def_seq.append(mov_ds) return def_seq
[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)