forked from IntrAnatSEEGSoftware/IntrAnat
-
Notifications
You must be signed in to change notification settings - Fork 0
/
scriptVep2IntranatCSV.py
156 lines (138 loc) · 6.5 KB
/
scriptVep2IntranatCSV.py
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
# ls /gin/data/database/03-preprocessed/Freesurfer/*/mri/aparc+aseg.vep.mgz
#
#
# 1. Identify VEP files from the DB
# 2. Check if there is already a copy in IntrAnat database
# 3. Import the missing ones in IntrAnat
# 4. Export CSV for each implantation
#
import subprocess
import glob
import sys
from brainvisa import axon
from brainvisa.data.readdiskitem import ReadDiskItem
from brainvisa.data.writediskitem import WriteDiskItem
from freesurfer.brainvisaFreesurfer import launchFreesurferCommand
from brainvisa.processes import defaultContext
from brainvisa.data import neuroHierarchy
from soma.qt_gui.qt_backend import QtGui, QtCore
from locateElectrodes import LocateElectrodes
from externalprocesses import createItemDirs
from batch_csv import generateCsv
import concurrent.futures
freesurferDB = "/host/media/odavid/FTract/data/database/03-preprocessed/Freesurfer"
brainvisaDB = "/host/media/odavid/FTract/data/database/03-preprocessed/Brainvisa"
def findT1pre(subject):
rT1BV = ReadDiskItem('Raw T1 MRI', 'BrainVISA volume formats', requiredAttributes={'subject': subject})
allT1 = list(rT1BV.findValues({}, None, False))
if not allT1:
return None
idxT1pre = [i for i in range(len(allT1)) if 'T1pre' in str(allT1[i])]
if not idxT1pre:
return None
if len(idxT1pre) > 1:
print("WARNING, found multiple T1pre for subject ", subject, " Returning the first one\n", repr(allT1))
diT1pre = allT1[idxT1pre[0]]
return diT1pre
def importFsAtlas(subject, proto='Epilepsy', imgName='VEP', imgPath="default.mgz", diT1pre=None, fileVoxelType=None):
"""
Import Freesurfer atlas volume in T1 referential
subject name, protocol, image name (e.g. VEP), MGZ image path, T1pre ReadDiskItem, fileVoxelType (S16 or S32)
"""
# If external call: Find T1pre of the subject
if not diT1pre:
diT1pre = findT1pre(subject)
if not diT1pre:
return [False, ['No T1pre found ' + repr(subject)]]
# Where to copy the new files
acq = str(diT1pre.attributes()['acquisition']).replace('T1', imgName + "-")
# Importing VEP atlas to BrainVISA database
wdi = WriteDiskItem('FreesurferAtlas', 'NIFTI-1 image')
di = wdi.findValue({'center': proto, 'acquisition': acq, 'subject': subject})
# Create the folder if doesn't exist
createItemDirs(di)
# Reslice volume to match T1pre
try:
launchFreesurferCommand(defaultContext(), None, 'mri_convert', '-i', imgPath,
'-o', str(di.fullPath()), '-rl', str(diT1pre.fullPath()),
'-rt', 'nearest', '-nc')
except:
print("ERROR: ", "Could not launch Freesurfer command")
return False, ["Could not launch Freesurfer command"]
# Convert to AIMS
if fileVoxelType is None:
fileVoxelType = 'S16' # Default value
if 'data_type' in di.attributes():
if di.attributes()['data_type'] in ['U32', 'S32']: # Do not convert to S16 if type is U32 or S32
fileVoxelType = di.attributes()['data_type']
ret = subprocess.call(['AimsFileConvert', '-i', str(di.fullPath()),
'-o', str(di.fullPath()), '-t', fileVoxelType])
# Add reference in the database (creates .minf)
neuroHierarchy.databases.insertDiskItem(di, update=True)
return True, []
def processVepFile(v):
veppath = v
v = v.replace(freesurferDB + "/", "").replace("/mri/aparc+aseg.vep.mgz", "")
v = v.replace('-','') # From 0013STA_13-01-2014 in Freesurfer to 0013STA_13012014 in Brainvisa
# 2) Check if it is already imported
# brainvisadb/Epilepsy/0001GRE_25112014/FreesurferAtlas/VEP-pre_2014-1-1/0001GRE_25112014-VEP-pre_2014-1-1.nii
vepBV = glob.glob(brainvisaDB + "/Epilepsy/" + v + "/FreesurferAtlas/VEP-pre_*/" + v + "-VEP-pre_*.nii")
if len(vepBV) > 0:
print("Already imported ", v, " -> ", repr(vepBV))
return (False, "Already imported "+ v+ " -> "+ repr(vepBV))
else:
print("Importing ", v)
# 3) Import it into BrainVisa
worked, msg = importFsAtlas(v, proto='Epilepsy', imgName='VEP', imgPath=veppath, fileVoxelType='S32')
if(worked):
print("Successfully imported ", v)
else:
print("Failed to import ", v, " -> ", msg)
return (False, "Failed to import " + v + " -> " + str(msg))
print("Exporting CSV")
isOk, errMsg = generateCsv(w, v, True, True)
if isOk:
print(v, ": CSV exported")
return (True, v + ": CSV exported")
else:
print(v, ": CSV export failed: ", errMsg)
return (False, v + ": CSV export failed: " + errMsg)
if __name__ == '__main__':
# Check if this is necessary to get ReadDiskItems
app = QtGui.QApplication(sys.argv)
axon.initializeProcesses()
w = LocateElectrodes(app=app, loadAll=True, isGui=False)
# Find available patients in BV database
rdi = ReadDiskItem( 'Subject', 'Directory',requiredAttributes={'_ontology':'brainvisa-3.2.0'}) #, requiredAttributes={'center':'Epilepsy'} )
w.allSubjects = list( rdi._findValues( {}, None, False ))
w.currentProtocol = 'Epilepsy'
w.subjects = [s.attributes()['subject'] for s in w.allSubjects if 'center' in s.attributes() and s.attributes()['center'] == w.currentProtocol]
w.subjects = sorted(w.subjects)
# 1) List subjects who have VEP files in Freesurfer database
vepFiles = glob.glob(freesurferDB + "/*/mri/aparc+aseg.vep.mgz")
print("Found ", len(vepFiles), " VEP files found")
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = []
for v in vepFiles:
futures.append(executor.submit(processVepFile,v))
for future in concurrent.futures.as_completed(futures):
print(future.result())
app.quit()
del app
#
# [
# "Compute MNI coordinates for all contacts",
# "Compute parcels for all contacts",
# "Compute MarsAtlas resection position",
# "Compute parcel metrics",
# "Save contact coordinates (.pts/.txt files)",
# "Save contact info (CSV file)",
# "Save contact info (BIDS .tsv)",
# "Save screenshots",
# "Save video (MP4)"],
# "Export", "Select options to run:",
# [True, True, False, False, True, True, self.bidspath is not None, False, False]
# selOptions = [True, True, True, False, True, True, False, False, False]
# locEl = LocateElectrodes(app=None, loadAll=False, isGui=False)
# locEl.loadPatient(v)
# locEl.exportAllWorker(selOptions)