import os import unittest from __main__ import vtk, qt, ctk, slicer from slicer.ScriptedLoadableModule import * import json import datetime import sys import nixModule import pathlib import chardet import re # # labkeySlicerPythonExtension # class imageBrowser(ScriptedLoadableModule): """Uses ScriptedLoadableModule base class, available at: https://github.com/Slicer/Slicer/blob/master/Base/Python/slicer/ScriptedLoadableModule.py """ def __init__(self, parent): ScriptedLoadableModule.__init__(self, parent) self.parent.title = "image Browser" # TODO make this more human readable by adding spaces self.parent.categories = ["dynamicSPECT"] self.parent.dependencies = [] self.parent.contributors = ["Andrej Studen (UL/FMF)"] # replace with "Firstname Lastname (Organization)" self.parent.helpText = """ Interface to irAEMM files in LabKey """ self.parent.acknowledgementText = """ Developed within the medical physics research programme of the Slovenian research agency. """ # replace with organization, grant and thanks. # # labkeySlicerPythonExtensionWidget # class imageBrowserWidget(ScriptedLoadableModuleWidget): """Uses ScriptedLoadableModuleWidget base class, available at: https://github.com/Slicer/Slicer/blob/master/Base/Python/slicer/ScriptedLoadableModule.py """ def setup(self): print("Setting up imageBrowserWidget") ScriptedLoadableModuleWidget.setup(self) # Instantiate and connect widgets ... self.logic=imageBrowserLogic(self) self.addInfoSection() self.addSetupSection() self.addPatientsSelector() self.addSegmentEditor() self.addWindowManipulator() def addInfoSection(self): #a python overview of json settings infoCollapsibleButton = ctk.ctkCollapsibleButton() infoCollapsibleButton.text = "Info" self.layout.addWidget(infoCollapsibleButton) infoLayout = qt.QFormLayout(infoCollapsibleButton) self.participantField=qt.QLabel("PatientId") infoLayout.addRow("Participant field:",self.participantField) self.ctField=qt.QLabel("ct") infoLayout.addRow("Data field (CT):",self.ctField) self.spectField=qt.QLabel("spect") infoLayout.addRow("Data field (SPECT):",self.spectField) self.userField=qt.QLabel("Loading") infoLayout.addRow("User",self.userField) self.idField=qt.QLabel("Loading") infoLayout.addRow("ID",self.idField) #Add logic at some point #self.logic=imageBrowserLogic(self) def addPatientsSelector(self): # # Patients Area # patientsCollapsibleButton = ctk.ctkCollapsibleButton() patientsCollapsibleButton.text = "Patients" #don't add it yet self.layout.addWidget(patientsCollapsibleButton) patientsFormLayout = qt.QFormLayout(patientsCollapsibleButton) self.patientList=qt.QComboBox() self.patientList.currentIndexChanged.connect(self.onPatientListChanged) self.patientList.setEditable(True) self.patientList.setInsertPolicy(qt.QComboBox.NoInsert) patientsFormLayout.addRow("Patient:",self.patientList) self.visitList=qt.QComboBox() self.visitList.currentIndexChanged.connect(self.onVisitListChanged) patientsFormLayout.addRow("Visit:",self.visitList) self.ctCode=qt.QLabel("ctCode") patientsFormLayout.addRow("CT:",self.ctCode) self.spectCode=qt.QLabel("spectCode") patientsFormLayout.addRow("SPECT:",self.spectCode) self.patientLoad=qt.QPushButton("Load") self.patientLoad.clicked.connect(self.onPatientLoadButtonClicked) patientsFormLayout.addRow("Load patient",self.patientLoad) self.patientSave=qt.QPushButton("Save") self.patientSave.clicked.connect(self.onPatientSaveButtonClicked) patientsFormLayout.addRow("Save segmentation",self.patientSave) self.patientClear=qt.QPushButton("Clear") self.patientClear.clicked.connect(self.onPatientClearButtonClicked) patientsFormLayout.addRow("Clear patient",self.patientClear) self.keepCached=qt.QCheckBox("keep Cached") self.keepCached.setChecked(1) patientsFormLayout.addRow("Keep cached",self.keepCached) self.forceReload=qt.QCheckBox("Force reload") self.forceReload.setChecked(0) patientsFormLayout.addRow("Force reload",self.forceReload) def addSetupSection(self): setupCollapsibleButton = ctk.ctkCollapsibleButton() setupCollapsibleButton.text = "Setup" self.layout.addWidget(setupCollapsibleButton) #Form layout (maybe one can think of more intuitive layouts) setupFormLayout = qt.QFormLayout(setupCollapsibleButton) self.serverList=qt.QComboBox() self.serverList.addItem('') #self.setupList.addItem("limfomiPET_iBrowser.json") #self.setupList.addItem("limfomiPET_iBrowser_selected.json") #self.setupList.addItem("iraemm_iBrowserProspective.json") #self.setupList.addItem("iraemm_iBrowserRetrospective.json") self.setupList.addItem("cardiacSPECT_iBrowser.json") self.setupList.currentIndexChanged.connect(self.onSetupListChanged) setupFormLayout.addRow("Setup:",self.setupList) def addSegmentEditor(self): editorCollapsibleButton = ctk.ctkCollapsibleButton() editorCollapsibleButton.text = "Segment Editor" self.layout.addWidget(editorCollapsibleButton) hLayout=qt.QVBoxLayout(editorCollapsibleButton) self.segmentEditorWidget=slicer.qMRMLSegmentEditorWidget() hLayout.addWidget(self.segmentEditorWidget) self.segmentEditorWidget.setMRMLScene(slicer.mrmlScene) segEditorNode=slicer.vtkMRMLSegmentEditorNode() slicer.mrmlScene.AddNode(segEditorNode) self.segmentEditorWidget.setMRMLSegmentEditorNode(segEditorNode) def addWindowManipulator(self): windowManipulatorCollapsibleButton=ctk.ctkCollapsibleButton() windowManipulatorCollapsibleButton.text="CT Window Manipulator" self.layout.addWidget(windowManipulatorCollapsibleButton) hLayout=qt.QHBoxLayout(windowManipulatorCollapsibleButton) ctWins={'CT:bone':self.onCtBoneButtonClicked, 'CT:air':self.onCtAirButtonClicked, 'CT:abdomen':self.onCtAbdomenButtonClicked, 'CT:brain':self.onCtBrainButtonClicked, 'CT:lung':self.onCtLungButtonClicked} for ctWin in ctWins: ctButton=qt.QPushButton(ctWin) ctButton.clicked.connect(ctWins[ctWin]) hLayout.addWidget(ctButton) def onSetupListChanged(self,i): status=self.logic.setConfig(self.setupList.currentText) try: if status['error']=='FILE NOT FOUND': print('File {} not found.'.format(self.setupList.currentText)) return except KeyError: pass #sort ids ids=status['ids'] ids.sort() self.updatePatientList(ids) self.onPatientListChanged(0) def onServerListChanged(self,i): status=self.logic.setServer(self.serverList.currentText) try: if status['error']=='KEY ERROR': self.serverList.setStyleSheet('background-color: violet') if status['error']=='ID ERROR': self.serverList.setStyleSheet('background-color: red') return except KeyError: pass self.idField.setText(status['id']) self.userField.setText(status['displayName']) self.serverList.setStyleSheet('background-color: green') def onPatientListChanged(self,i): dt=datetime.datetime self.visitList.clear() self.spectCode.setText("") self.ctCode.setText("") #add potential filters from setup to dbFilter ds=self.logic.getDataset(dbFilter={'participant':self.patientList.currentText}) visitVar=self.logic.getVarName(var='visitField') studyDateVar=self.logic.getVarName(var='studyDateField') #label is a combination of sequence number and date of imaging try: seq={row['SequenceNum']: {'label':row[visitVar],'date': parseDate(row[studyDateVar])} for row in ds['rows']} except TypeError: #if studyDate is empty, this will return no possible visits return #apply lookup to visitVar if available try: seq={x: {'label':ds['lookups'][visitVar][seq[x]['label']], 'date':seq[x]['date']} for x in seq} except KeyError: pass #format label seq={x:'{} ({})'.format(seq[x]['label'],dt.strftime(seq[x]['date'],'%d-%b-%Y')) for x in seq} for s in seq: #onVisitListChanged is called for every addItem self.visitList.addItem(seq[s],s) #self.onVisitListChanged(0) def onVisitListChanged(self,i): #ignore calls on empty list if self.visitList.count==0: return #get sequence num s=self.visitList.itemData(i) print("Visit: SequenceNum:{}, label{}".format(s,self.visitList.currentText)) dbFilter={'participant':self.patientList.currentText, 'seqNum':s} ds=self.logic.getDataset(dbFilter=dbFilter) if not len(ds['rows'])==1: print("Found incorrect number {} of matches for [{}]/[{}]".\ format(len(ds['rows']),\ self.patientList.currentText,s)) row=ds['rows'][0] #copy row properties for data access self.currentRow=row self.spectCode.setText(row[self.spectField.text]) self.ctCode.setText(row[self.ctField.text]) #self.segmentationCode.setText(row[self.segmentationField.text]) def updatePatientList(self,ids): self.patientList.clear() for id in ids: self.patientList.addItem(id) def onPatientLoadButtonClicked(self): print("Load") #delegate loading to logic self.logic.loadImages(self.currentRow,self.keepCached.isChecked(), self.forceReload.isChecked()) self.logic.loadSegmentation(self.currentRow) self.setSegmentEditor() #self.logic.loadReview(self.currentRow) #self.logic.loadAE(self.currentRow) #self.onReviewSegmentChanged() def setSegmentEditor(self): #use current row to set segment in segment editor self.segmentEditorWidget.setSegmentationNode( self.logic.volumeNode['Segmentation']) self.segmentEditorWidget.setMasterVolumeNode( self.logic.volumeNode['SPECT']) def onReviewSegmentChanged(self): pass def onPatientClearButtonClicked(self): self.logic.clearVolumesAndSegmentations() self.patientSave.setStyleSheet('background-color:gray') def onPatientSaveButtonClicked(self): status=self.logic.saveSegmentation() if status: self.patientSave.setStyleSheet('background-color:green') else: self.patientSave.setStyleSheet('background-color:red') def onCtBoneButtonClicked(self): self.logic.setWindow('CT:bone') def onCtAirButtonClicked(self): self.logic.setWindow('CT:air') def onCtAbdomenButtonClicked(self): self.logic.setWindow('CT:abdomen') def onCtBrainButtonClicked(self): self.logic.setWindow('CT:brain') def onCtLungButtonClicked(self): self.logic.setWindow('CT:lung') def cleanup(self): pass def loadLibrary(name): #utility function to load nix library from git fwrapper=nixModule.getWrapper('nixWrapper.py') p=pathlib.Path(fwrapper) sys.path.append(str(p.parent)) import nixWrapper return nixWrapper.loadLibrary(name) def parseDate(x): print(f'Parsing date from {x}') dt=datetime.datetime try: return dt.strptime(x,'%Y/%m/%d %H:%M:%S.%f') except ValueError: pass try: return dt.strptime(x,'%Y-%m-%d %H:%M:%S.%f') except ValueError: pass #raise TypeError(f'Could not convert date {x}') return dt.fromtimestamp(0) # # imageBrowserLogic # class imageBrowserLogic(ScriptedLoadableModuleLogic): """This class should implement all the actual computation done by your module. The interface should be such that other python code can import this class and make use of the functionality without requiring an instance of the Widget. Uses ScriptedLoadableModuleLogic base class, available at: https://github.com/Slicer/Slicer/blob/master/Base/Python/slicer/ScriptedLoadableModule.py """ def __init__(self,parent=None): ScriptedLoadableModuleLogic.__init__(self, parent) print('imageBrowserLogic loading') if not parent==None: #use layout and data from parent widget self.parent=parent fhome=os.path.expanduser('~') fsetup=os.path.join(fhome,'.labkey','setup.json') try: with open(fsetup) as f: self.setup=json.load(f) except FileNotFoundError: self.setup={} try: pt=self.setup['paths'] except KeyError: self.setup['paths']={} lName='labkeyInterface' loadLibrary(lName) import labkeyInterface import labkeyDatabaseBrowser import labkeyFileBrowser self.network=labkeyInterface.labkeyInterface() self.dbBrowser=labkeyDatabaseBrowser self.fBrowser=labkeyFileBrowser self.tempDir=os.path.join(os.path.expanduser('~'),'temp') if not os.path.isdir(self.tempDir): os.mkdir(self.tempDir) self.lookups={} #self.segmentList=["Liver","Blood","Marrow","Disease","Deauville"] self.segmentList=["One","Two","Three","Four","Five","Six","Seven","Eight"] print('imageBrowserLogic setup complete') def setServer(self,serverName): #additional way of setting the labkey network interface #if no parent was provided in logic initialization (stand-alone mode) status={} fileName="NONE" if serverName=="astuden": fileName="astuden.json" if serverName=="llezaic": fileName="llezaic.json" if fileName=="NONE": print("No path was associated with server {}".format(serverName)) status['error']='KEY ERROR' return status fconfig=os.path.join(os.path.expanduser('~'),'.labkey',fileName) self.network.init(fconfig) self.remoteId=self.network.getUserId() if self.remoteId==None: status['error']='ID ERROR' return status status['displayName']=self.remoteId['displayName'] status['id']=self.remoteId['id'] #reset db and fb (they are thin classes anyhow) self.db=self.dbBrowser.labkeyDB(self.network) self.fb=self.fBrowser.labkeyFileBrowser(self.network) return status def setConfig(self,configName): status={} fileName=os.path.join(os.path.expanduser('~'),'.labkey',configName) if not os.path.isfile(fileName): status['error']='FILE NOT FOUND' return status with open(fileName,'r') as f: self.isetup=json.load(f) #self.project=self.isetup['project'] #"iPNUMMretro/Study" #self.schema='study' #self.dataset=self.isetup['query'] #include filters... ds=self.getDataset() try: filterValue=self.isetup['filterEntries'] except KeyError: #this is default ids=[row[self.isetup['participantField']] for row in ds['rows']] status['ids']=list(set(ids)) return status #look for entries where segmentation was already done dsSet=self.getDataset('SegmentationsMaster') segMap={'{}:{}'.format(r['patientCode'],r['visitCode']):r['comments'] for r in dsSet['rows']} ids=[] for r in ds['rows']: code='{}:{}'.format(r['patientCode'],r['visitCode']) try: comment=segMap[code] except KeyError: ids.append(r['patientCode']) continue if comment==filterValue: ids.append(r['patientCode']) status['ids']=list(set(ids)) return status def getVarName(self,name="Imaging",var="visitField"): dset=self.isetup['datasets'][name] defaults={"visitField":"imagingVisitId","studyDateField":"studyDate"} try: return dset[var] except KeyError: return defaults[var] def getDataset(self,name="Imaging",dbFilter={}): dset=self.isetup['datasets'][name] project=dset['project'] schema=dset['schema'] query=dset['query'] #add default filters qFilter=[] try: for qf in dset['filter']: v=dset['filter'][qf] qFilter.append({'variable':qf,'value':v,'oper':'eq'}) except KeyError: pass for f in dbFilter: if f=='participant': qFilter.append({'variable':self.isetup['participantField'], 'value':dbFilter[f],'oper':'eq'}) continue if f=='seqNum': qFilter.append({'variable':'SequenceNum', 'value':'{}'.format(dbFilter[f]), 'oper':'eq'}) continue qFilter.append({'variable':f,'value':dbFilter[f],'oper':'eq'}) try: ds=self.db.selectRows(project,schema,query, \ qFilter,dset['view']) except KeyError: ds=self.db.selectRows(project,schema,query,qFilter) #get lookups as well lookups={} for f in ds['metaData']['fields']: try: lookup=f['lookup'] except KeyError: continue var=f['name'] lookupCode='{}:{}'.format(lookup['schema'],lookup['queryName']) try: lookups[var]=self.lookups[lookupCode] except KeyError: self.lookups[lookupCode]=self.loadLookup(project,lookup) lookups[var]=self.lookups[lookupCode] return {'rows':ds['rows'],'lookups':lookups} def loadLookup(self,project,lookup): qData={} key=lookup['keyColumn'] value=lookup['displayColumn'] fSet=self.db.selectRows(project,lookup['schema'],lookup['queryName'],[]) for q in fSet['rows']: qData[q[key]]=q[value] return qData def loadImage(self,iData): #unpack iData idx=iData['idx'] path=iData['path'] keepCached=iData['keepCached'] try: forceReload=iData['forceReload'] except KeyError: forceReload=False dset=self.isetup['datasets'][iData['dataset']] localPath=os.path.join(self.tempDir,path[-1]) if not os.path.isfile(localPath) or forceReload: #download from server remotePath=self.fb.formatPathURL(dset['project'],'/'.join(path)) if not self.fb.entryExists(remotePath): print("Failed to get {}".format(remotePath)) return #overwrites existing file from remote self.fb.readFileToFile(remotePath,localPath) properties={} filetype='VolumeFile' #make sure segmentation gets loaded as a labelmap if idx=="Segmentation": filetype='SegmentationFile' #properties["labelmap"]=1 self.volumeNode[idx]=slicer.util.loadNodeFromFile(localPath, filetype=filetype,properties=properties) if not keepCached: pass #os.remove(localPath) def loadImages(self,row,keepCached, forceReload=False): #fields={'ctResampled':True,'petResampled':False} fields={"CT":self.parent.ctField.text,\ "SPECT":self.parent.spectField.text} path=[self.isetup['imageDir'],row['patientCode'],row['visitCode']] relativePaths={x:path+[row[y]] for (x,y) in fields.items()} self.volumeNode={} for f in relativePaths: iData={'idx':f,'path':relativePaths[f], 'keepCached':keepCached,'dataset':'Imaging', 'forceReload':forceReload} self.loadImage(iData) #mimic abdominalCT standardized window setting self.volumeNode['CT'].GetScalarVolumeDisplayNode().\ SetAutoWindowLevel(False) self.volumeNode['CT'].GetScalarVolumeDisplayNode().\ SetWindowLevel(1400, -500) #set colormap for PET to PET-Heat (this is a verbatim setting from #the Volumes->Display->Lookup Table colormap identifier) self.volumeNode['SPECT'].GetScalarVolumeDisplayNode().\ SetAndObserveColorNodeID(\ slicer.util.getNode('Inferno').GetID()) slicer.util.setSliceViewerLayers(background=self.volumeNode['CT'],\ foreground=self.volumeNode['SPECT'],foregroundOpacity=0.5,fit=True) def loadSegmentation(self,row, loadFile=1): dbFilter={'User':'{}'.format(self.remoteId['id']), 'participant':row[self.isetup['participantField']], 'visitCode':row['visitCode']} ds=self.getDataset(name='SegmentationsMaster', dbFilter=dbFilter) if len(ds['rows'])>1: print('Multiple segmentations found!') return if len(ds['rows'])==1: #update self.segmentationEntry self.segmentationEntry=ds['rows'][0] self.segmentationEntry['origin']='database' if loadFile: self.loadSegmentationFromEntry() return #create new segmentation self.createSegmentation(row) def getSegmentationPath(self): path=[self.isetup['imageDir'], self.segmentationEntry['patientCode'], self.segmentationEntry['visitCode']] path.append('Segmentations') return path def loadSegmentationFromEntry(self): #compile path entry=self.segmentationEntry path=self.getSegmentationPath() path.append(entry['latestFile']) iData={'idx':'Segmentation','path':path, 'keepCached':1,'dataset':'SegmentationsMaster'} self.loadImage(iData) #look for missing segments segNode=self.volumeNode['Segmentation'] seg=segNode.GetSegmentation() segNames=[seg.GetNthSegmentID(i) for i in range(seg.GetNumberOfSegments())] print('Segments') try: segmentList=self.isetup['segmentList'] except KeyError: segmentList=self.segmentList for s in segmentList: if s not in segNames: seg.AddEmptySegment(s,s) print(s) print('Done') def saveSegmentation(self): #get the latest key by adding an entry to SegmentationList copyFields=[self.isetup['participantField'],'patientCode','visitCode','User'] outRow={x:self.segmentationEntry[x] for x in copyFields} sList=self.isetup['datasets']['SegmentationsList'] resp=self.db.modifyRows('insert',sList['project'], sList['schema'],sList['query'],[outRow]) encoding=chardet.detect(resp)['encoding'] respJSON=json.loads(resp.decode(encoding)) outRow=respJSON['rows'][0] #print(outRow) #construct file name with known key uName=re.sub(' ','_',self.remoteId['displayName']) fName='Segmentation_{}-{}_{}_{}.nrrd'.format( self.segmentationEntry['patientCode'], self.segmentationEntry['visitCode'], uName,outRow['Key']) path=self.getSegmentationPath() path.append(fName) status=self.saveNode(self.volumeNode['Segmentation'],'SegmentationsMaster',path) #update SegmentationList with know file name outRow['Segmentation']=fName self.db.modifyRows('update',sList['project'], sList['schema'],sList['query'],[outRow]) #update SegmentationsMaster self.segmentationEntry['latestFile']=fName self.segmentationEntry['version']=outRow['Key'] des=self.isetup['datasets']['SegmentationsMaster'] op='insert' if self.segmentationEntry['origin']=='database': op='update' print('Saving: mode={}'.format(op)) resp=self.db.modifyRows(op,des['project'], des['schema'],des['query'],[self.segmentationEntry]) print(resp) #since we loaded a version, origin should be set to database self.loadSegmentation(self.segmentationEntry,0) return status #self.segmentationEntry['origin']='database' def saveNode(self,node,dataset,path): fName=path[-1] localPath=os.path.join(self.tempDir,fName) slicer.util.saveNode(node,localPath) dset=self.isetup['datasets'][dataset] #exclude file name when building directory structure remotePath=self.fb.buildPathURL(dset['project'],path[:-1]) remotePath+='/'+fName self.fb.writeFileToFile(localPath,remotePath) return self.fb.entryExists(remotePath) #add entry to segmentation list def createSegmentation(self,entry): #create segmentation entry for database update #note that origin is not set to database copyFields=[self.isetup['participantField'],'patientCode','visitCode','SequenceNum'] #copyFields=['ParticipantId','patientCode','visitCode','SequenceNum'] self.segmentationEntry={x:entry[x] for x in copyFields} self.segmentationEntry['User']=self.remoteId['id'] self.segmentationEntry['origin']='created' self.segmentationEntry['version']=-1111 #create a segmentation node uName=re.sub(' ','_',self.remoteId['displayName']) segNode=slicer.vtkMRMLSegmentationNode() self.volumeNode['Segmentation']=segNode segNode.SetName('Segmentation_{}_{}_{}'. format(entry['patientCode'],entry['visitCode'],uName)) slicer.mrmlScene.AddNode(segNode) segNode.CreateDefaultDisplayNodes() segNode.SetReferenceImageGeometryParameterFromVolumeNode(self.volumeNode['SPECT']) try: segmentList=self.isetup['segmentList'] except KeyError: segmentList=self.segmentList for s in segmentList: segNode.GetSegmentation().AddEmptySegment(s,s) def clearVolumesAndSegmentations(self): nodes=slicer.util.getNodesByClass("vtkMRMLVolumeNode") nodes.extend(slicer.util.getNodesByClass("vtkMRMLSegmentationNode")) res=[slicer.mrmlScene.RemoveNode(f) for f in nodes] #self.segmentationNode=None #self.reviewResult={} #self.aeList={} def setWindow(self,windowName): #default w=1400 l=-500 if windowName=='CT:bone': w=1000 l=400 if windowName=='CT:air': w=1000 l=-426 if windowName=='CT:abdomen': w=350 l=40 if windowName=='CT:brain': w=100 l=50 if windowName=='CT:lung': w=1400 l=-500 self.volumeNode['CT'].GetScalarVolumeDisplayNode().\ SetWindowLevel(w,l) print('setWindow: {} {}/{}'.format(windowName,w,l)) class imageBrowserTest(ScriptedLoadableModuleTest): """ This is the test case for your scripted module. Uses ScriptedLoadableModuleTest base class, available at: https://github.com/Slicer/Slicer/blob/master/Base/Python/slicer/ScriptedLoadableModule.py """ def setup(self): """ Do whatever is needed to reset the state - typically a scene clear will be enough. """ slicer.mrmlScene.Clear(0) def runTest(self): """Run as few or as many tests as needed here. """ self.setUp() self.test_irAEMMBrowser() def test_irAEMMBrowser(self): """ Ideally you should have several levels of tests. At the lowest level tests sould exercise the functionality of the logic with different inputs (both valid and invalid). At higher levels your tests should emulate the way the user would interact with your code and confirm that it still works the way you intended. One of the most important features of the tests is that it should alert other developers when their changes will have an impact on the behavior of your module. For example, if a developer removes a feature that you depend on, your test should break so they know that the feature is needed. """ self.delayDisplay("Starting the test") # # first, get some data #