import slicer import os import subprocess import re import dicom dicomModify=os.getenv("HOME") if not dicomModify==None: dicomModify+="/software/install/" dicomModify+="dicomModify/bin/dicomModify" class loadDicom(slicer.ScriptedLoadableModule.ScriptedLoadableModule): def __init__(self,parent): slicer.ScriptedLoadableModule.ScriptedLoadableModule.__init__(self, parent) self.className="loadDicom" self.parent.title="loadDicom" self.parent.categories = ["LabKey"] self.parent.dependencies = [] self.parent.contributors = ["Andrej Studen (UL/FMF)"] # replace with "Firstname Lastname (Organization)" self.parent.helpText = """ utilities for parsing dicom entries """ self.parent.acknowledgementText = """ Developed within the medical physics research programme of the Slovenian research agency. """ # replace with organization, grant and thanks. class loadDicomWidget(slicer.ScriptedLoadableModule.ScriptedLoadableModuleWidget): """Uses ScriptedLoadableModuleWidget base class, available at: https://github.com/Slicer/Slicer/blob/master/Base/Python/slicer/ScriptedLoadableModule.py """ def setup(self): slicer.ScriptedLoadableModule.ScriptedLoadableModuleWidget.setup(self) class loadDicomLogic(slicer.ScriptedLoadableModule.ScriptedLoadableModuleLogic): def __init__(self,parent): slicer.ScriptedLoadableModule.ScriptedLoadableModuleLogic.__init__(self, parent) self.tag={ 'studyDate': {'tag':"0008,0020",'VR':'DA'}, 'studyTime': {'tag':"0008,0030",'VR':'TM'}, 'seriesTime': {'tag':"0008,0031",'VR':'TM'}, 'modality': {'tag':"0008,0060",'VR':'CS'}, 'presentationIntentType': {'tag':"0008,0068",'VR':'CS'}, 'manufacturer': {'tag':"0008,0070",'VR':'LO'}, 'institutionName': {'tag':"0008,0080",'VR':'LO'}, 'studyDescription': {'tag':"0008,1030",'VR':'LO'}, 'seriesDescription': {'tag':"0008,103e",'VR':'LO'}, 'manufacturerModelName': {'tag':"0008,1090",'VR':'LO'}, 'patientName': {'tag':"0010,0010",'VR':'PN'}, 'patientId': {'tag':"0010,0020",'VR':'LO'}, 'patientBirthDate': {'tag':"0010,0030",'VR':'DA'}, 'patientSex': {'tag':"0010,0040",'VR':'CS'}, 'patientAge': {'tag':"0010,1010",'VR':'AS'}, 'patientComments': {'tag':"0010,4000",'VR':'LT'}, 'sequenceName': {'tag':"0018,0024",'VR':'SH'}, 'kVP': {'tag':"0018,0060",'VR':'DS'}, 'percentPhaseFieldOfView': {'tag':"0018,0094",'VR':'DS'}, 'xRayTubeCurrent': {'tag':"0018,1151",'VR':'IS'}, 'exposure': {'tag':"0018,1152",'VR':'IS'}, 'imagerPixelSpacing': {'tag':"0018,1164",'VR':'DS'}, 'bodyPartThickness': {'tag':"0018,11a0",'VR':'DS'}, 'compressionForce': {'tag':"0018,11a2",'VR':'DS'}, 'viewPosition': {'tag':"0018,5101",'VR':'CS'}, 'fieldOfViewHorizontalFlip': {'tag':"0018,7034",'VR':'CS'}, 'studyInstanceUid': {'tag':"0020,000d",'VR':'UI'}, 'seriesInstanceUid': {'tag':"0020,000e",'VR':'UI'}, 'studyId': {'tag':"0020,0010",'VR':'SH'}, 'seriesNumber': {'tag':"0020,0011",'VR':'IS'}, 'instanceNumber': {'tag':"0020,0013",'VR':'IS'}, 'frameOfReferenceInstanceUid': {'tag':"0020,0052",'VR':'UI'}, 'imageLaterality': {'tag':"0020,0062",'VR':'CS'}, 'imagesInAcquisition': {'tag':"0020,1002",'VR':'IS'}, 'photometricInterpretation': {'tag':"0028,0004",'VR':'CS'}, 'reconstructionMethod': {'tag':"0054,1103",'VR':'LO'} } self.tagPyDicom={ 'studyDate': 0x00080020, 'studyTime': 0x00080030, 'modality': 0x00080060, 'presentationIntentType': 0x00080068, 'manufacturer': 0x00080070, 'studyDescription': 0x00081030, 'seriesDescription': 0x0008103e, 'patientName': 0x00100010, 'patientId': 0x00100020, 'patientBirthDate': 0x00100030, 'patientSex': 0x00100040, 'patientComments': 0x00104000, 'sequenceName': 0x00180024, 'kVP': 0x00180060, 'percentPhaseFieldOfView': 0x00180094, 'xRayTubeCurrent': 0x00181151, 'exposure': 0x00181152, 'imagerPixelSpacing': 0x00181164, 'bodyPartThickness': 0x001811a0, 'compressionForce': 0x001811a2, 'viewPosition': 0x00185101, 'studyInstanceUid': 0x0020000d, 'seriesInstanceUid': 0x0020000e, 'studyId': 0x00200010, 'seriesNumber': 0x00200011, 'instanceNumber': 0x00200013, 'frameOfReferenceInstanceUid': 0x00200052 } #new_dict_items={ # 0x001811a0: ('DS','BodyPartThickness','Body Part Thickness') #} #dicom.datadict.add_dict_entries(new_dict_items) def getHex(self,key): #convert string to hex key; fv=key.split(",") return int(fv[0],16)*0x10000+int(fv[1],16) def load(self,sNet,dir,doRemove=True): print("Loading dir {}").format(dir) dicomFiles=sNet.listRelativeDir(dir) #filelist=[os.path.join(dir,f) for f in os.listdir(dir)] filelist=[] for f in dicomFiles: localPath=sNet.DownloadFileToCache(f) f0=localPath f1=f0+"1" if not dicomModify==None: subprocess.call(dicomModify+" "+f0+" "+f1+" && mv "+f1+" "+f0+";", shell=True) filelist.append(localPath) try: loadables=self.volumePlugin.examineForImport([filelist]) except AttributeError: self.volumePlugin=slicer.modules.dicomPlugins['DICOMScalarVolumePlugin']() loadables=self.volumePlugin.examineForImport([filelist]) for loadable in loadables: #TODO check if it makes sense to load a particular loadable fileBuffer = open(loadable.files[0]) plan=dicom.read_file(fileBuffer) if loadable.name.find('imageOrientationPatient')>-1: continue volumeNode=self.volumePlugin.load(loadable) if volumeNode != None: vName='Series'+plan[0x00200011].value volumeNode.SetName(vName) try: loadableRTs=self.RTPlugin.examineForImport([filelist]) except: self.RTPlugin=plugin=slicer.modules.dicomPlugins['DicomRtImportExportPlugin']() loadableRTs=self.RTPlugin.examineForImport([filelist]) for loadable in loadableRTs: segmentationNode=self.RTPlugin.load(loadable) #if segmentationNode!= None: # segmentationNode.SetName('SegmentationBR') if not doRemove: return for f in filelist: os.remove(f) def applyFilter(self,loadable,filter,nodeMetadata): filterOK=True print("Opening {}").format(loadable.files[0]); fileBuffer = open(loadable.files[0]) try: plan = dicom.read_file(fileBuffer) except: return False for key in filter: #try: # fileValue=plan[self.tag[key]].value #except KeyError: # fileValue=None fileValue=dicomValue(loadable.files[0],self.tag[key]['tag']) if filter[key]=="SeriesLabel": nodeMetadata['seriesLabel']=fileValue continue if not filter[key]==None: if not fileValue==filter[key]: print("File {} failed for tag {}: {}/{}").format( loadable.files[0],key,fileValue,filter[key]) filterOK=False break nodeMetadata[key]=fileValue return filterOK def loadVolumes(self,sNet,dir,filter,doRemove=True): #returns all series from the directory, each as a separate node in a node list #filter is a dictionary of speciifed dicom values, if filter(key)=None, that values #get set, if it isn't, the file gets checked for a match print("Loading dir {}").format(dir) dicomFiles=sNet.listRelativeDir(dir) #filelist=[os.path.join(dir,f) for f in os.listdir(dir)] filelist=[] for f in dicomFiles: localPath=sNet.DownloadFileToCache(f) f0=localPath f1=f0+"1" if not dicomModify==None: subprocess.call(dicomModify+" "+f0+" "+f1+" && mv "+f1+" "+f0+";", shell=True) filelist.append(localPath) try: loadables=self.volumePlugin.examineForImport([filelist]) except AttributeError: self.volumePlugin=slicer.modules.dicomPlugins['DICOMScalarVolumePlugin']() loadables=self.volumePlugin.examineForImport([filelist]) volumeNodes=[] print("Number of loadables:{}").format(len(loadables)) for loadable in loadables: #TODO check if it makes sense to load a particular loadable print "Loading {} number of files: {}".format(loadable.name,len(loadable.files)) for f in loadable.files: print "\t {}".format(f) #perform checks nodeMetadata={} filterOK=self.applyFilter(loadable,filter,nodeMetadata) if not filterOK: #skip this loadable continue volumeNode=self.volumePlugin.load(loadable,"DCMTK") if volumeNode != None: vName='Series'+nodeMetadata['seriesLabel'] volumeNode.SetName(vName) volume={'node':volumeNode,'metadata':nodeMetadata} volumeNodes.append(volume) if doRemove: for f in filelist: os.remove(f) return volumeNodes def loadSegmentations(self,net,dir,filter,doRemove=True): print("Loading dir {}").format(dir) dicomFiles=net.listRelativeDir(dir) filelist=[net.DownloadFileToCache(f) for f in dicomFiles] segmentationNodes=[] try: loadableRTs=self.RTPlugin.examineForImport([filelist]) except: self.RTPlugin=plugin=slicer.modules.dicomPlugins['DicomRtImportExportPlugin']() loadableRTs=self.RTPlugin.examineForImport([filelist]) for loadable in loadableRTs: nodeMetadata={} filterOK=self.applyFilter(loadable,filter,nodeMetadata) if not filterOK: continue success=self.RTPlugin.load(loadable) if not success: print("Could not load RT structure set") return segNodes=slicer.util.getNodesByClass("vtkMRMLSegmentationNode") segmentationNode=segNodes[0] #assume we loaded the first node in list if segmentationNode != None: sName='Segmentation'+nodeMetadata['seriesLabel'] segmentationNode.SetName(sName) segmentation={'node':segmentationNode,'metadata':nodeMetadata} segmentationNodes.append(segmentation) if not doRemove: return segmentationNodes for f in filelist: os.remove(f) return segmentationNodes def isDicom(file): try: f=open(file,'rb') except IOError: return False f.read(128) dt=f.read(4) f.close() return dt=='DICM' def dicomValue(file,tag): dcmdump=os.path.join(os.environ['SLICER_HOME'],"bin") dcmdump=os.path.join(dcmdump,"dcmdump") try: out=subprocess.check_output([dcmdump,'+p','+P',tag,file]) print("Tag {} Line '{}'").format(tag,out) if len(out)==0: return out tag1="^\({}\)".format(tag) lst=out.split('\n') rpl=[re.sub(r'^.*\[(.*)\].*$',r'\1',f) for f in lst] mtch=[re.match(tag1,f) for f in lst] out=[x for y,x in zip(mtch,rpl) if not y==None] out=out[0] #out=re.sub(r'^.*\[(.*)\].*\n$',r'\1',out) #separate out series print("Tag {} Parsed value {}").format(tag,out) if out.find('\\')>-1: out=out.split('\\') return out except subprocess.CalledProcessError as e: return None def clearNodes(): nodes=[] nodes.extend(slicer.util.getNodesByClass("vtkMRMLScalarVolumeNode")) nodes.extend(slicer.util.getNodesByClass("vtkMRMLScalarVolumeDisplayNode")) nodes.extend(slicer.util.getNodesByClass("vtkMRMLSegmentationNode")) nodes.extend(slicer.util.getNodesByClass("vtkMRMLSegmentationDisplayNode")) for node in nodes: slicer.mrmlScene.RemoveNode(node)