Coverage for ibllib/pipes/ephys_tasks.py: 59%

1import logging (empty)

2import traceback (empty)

3from pathlib import Path (empty)

4import subprocess (empty)

5import re (empty)

6import shutil (empty)

7 

8import packaging.version (empty)

9import numpy as np (empty)

10import pandas as pd (empty)

11import spikeglx (empty)

12import neuropixel (empty)

13from ibldsp.utils import rms (empty)

14from ibldsp.waveform_extraction import extract_wfs_cbin (empty)

15import one.alf.io as alfio (empty)

16 

17from ibllib.misc import check_nvidia_driver (empty)

18from ibllib.pipes import base_tasks (empty)

19from ibllib.pipes.sync_tasks import SyncPulses (empty)

20from ibllib.ephys import ephysqc (empty)

21import ibllib.ephys.spikes (empty)

22from ibllib.qc.alignment_qc import get_aligned_channels (empty)

23from ibllib.plots.figures import LfpPlots, ApPlots, BadChannelsAp (empty)

24from ibllib.plots.figures import SpikeSorting as SpikeSortingPlots (empty)

25from ibllib.io.extractors.ephys_fpga import extract_sync (empty)

26from ibllib.ephys.spikes import sync_probes (empty)

27 

28 

29_logger = logging.getLogger("ibllib") (empty)

30 

31 

32class EphysRegisterRaw(base_tasks.DynamicTask): (empty)

33 """ 

34 Creates the probe insertions and uploads the probe descriptions file, also compresses the nidq files and uploads 

35 """ 

36 

37 priority = 100 (empty)

38 job_size = 'small' (empty)

39 

40 @property (empty)

41 def signature(self): (empty)

42 signature = { 7 ctx1aghideu

43 'input_files': [], 

44 'output_files': [('probes.description.json', 'alf', True)] 

45 } 

46 return signature 7 ctx1aghideu

47 

48 def _run(self): (empty)

49 

50 out_files = ibllib.ephys.spikes.probes_description(self.session_path, self.one) 1 ctx1u

51 

52 return out_files 1 ctx1u

53 

54 

55class EphysSyncRegisterRaw(base_tasks.DynamicTask): (empty)

56 """ 

57 Task to rename, compress and register raw daq data with .bin format collected using NIDAQ 

58 """ 

59 

60 priority = 90 (empty)

61 cpu = 2 (empty)

62 io_charge = 30 # this jobs reads raw ap files (empty)

63 job_size = 'small' (empty)

64 

65 @property (empty)

66 def signature(self): (empty)

67 signature = { 9 ctx1aghidejks

68 'input_files': [('*.*bin', self.sync_collection, True), 

69 ('*.meta', self.sync_collection, True), 

70 ('*.wiring.json', self.sync_collection, True)], 

71 'output_files': [('*nidq.cbin', self.sync_collection, True), 

72 ('*nidq.ch', self.sync_collection, True), 

73 ('*nidq.meta', self.sync_collection, True), 

74 ('*nidq.wiring.json', self.sync_collection, True)] 

75 } 

76 return signature 9 ctx1aghidejks

77 

78 def _run(self): (empty)

79 

80 out_files = [] 3 ctx1jks

81 

82 # Detect the wiring file 

83 wiring_file = next(self.session_path.joinpath(self.sync_collection).glob('*.wiring.json'), None) 3 ctx1jks

84 if wiring_file is not None: 3 ctx1jks

85 out_files.append(wiring_file) 3 ctx1jks

86 

87 # Search for .bin files in the sync_collection folder 

88 files = list(self.session_path.joinpath(self.sync_collection).glob('*nidq.*bin')) 3 ctx1jks

89 bin_file = files[0] if len(files) == 1 else None 3 ctx1jks

90 

91 # If we don't have a .bin/ .cbin file anymore see if we can still find the .ch and .meta files 

92 if bin_file is None: 3 ctx1jks

93 for ext in ['ch', 'meta']: 1 ctx1s

94 files = list(self.session_path.joinpath(self.sync_collection).glob(f'*nidq.{ext}')) 1 ctx1s

95 ext_file = files[0] if len(files) == 1 else None 1 ctx1s

96 if ext_file is not None: 1 ctx1s

97 out_files.append(ext_file) 1 ctx1s

98 

99 return out_files if len(out_files) > 0 else None 1 ctx1s

100 

101 # If we do find the .bin file, compress files (only if they haven't already been compressed) 

102 sr = spikeglx.Reader(bin_file) 2 ctx1jk

103 if sr.is_mtscomp: 2 ctx1jk

104 sr.close() 1 ctx1j

105 cbin_file = bin_file 1 ctx1j

106 assert cbin_file.suffix == '.cbin' 1 ctx1j

107 else: 

108 cbin_file = sr.compress_file() 1 ctx1k

109 sr.close() 1 ctx1k

110 bin_file.unlink() 1 ctx1k

111 

112 meta_file = cbin_file.with_suffix('.meta') 2 ctx1jk

113 ch_file = cbin_file.with_suffix('.ch') 2 ctx1jk

114 

115 out_files.append(cbin_file) 2 ctx1jk

116 out_files.append(ch_file) 2 ctx1jk

117 out_files.append(meta_file) 2 ctx1jk

118 

119 return out_files 2 ctx1jk

120 

121 

122class EphysCompressNP1(base_tasks.EphysTask): (empty)

123 priority = 90 (empty)

124 cpu = 2 (empty)

125 io_charge = 100 # this jobs reads raw ap files (empty)

126 job_size = 'small' (empty)

127 

128 @property (empty)

129 def signature(self): (empty)

130 signature = { 7 ctx1aghiqfr

131 'input_files': [('*ap.meta', f'{self.device_collection}/{self.pname}', True), 

132 ('*ap.*bin', f'{self.device_collection}/{self.pname}', True), 

133 ('*lf.meta', f'{self.device_collection}/{self.pname}', True), 

134 ('*lf.*bin', f'{self.device_collection}/{self.pname}', True), 

135 ('*wiring.json', f'{self.device_collection}/{self.pname}', False)], 

136 'output_files': [('*ap.meta', f'{self.device_collection}/{self.pname}', True), 

137 ('*ap.cbin', f'{self.device_collection}/{self.pname}', True), 

138 ('*ap.ch', f'{self.device_collection}/{self.pname}', True), 

139 ('*lf.meta', f'{self.device_collection}/{self.pname}', True), 

140 ('*lf.cbin', f'{self.device_collection}/{self.pname}', True), 

141 ('*lf.ch', f'{self.device_collection}/{self.pname}', True), 

142 ('*wiring.json', f'{self.device_collection}/{self.pname}', False)] 

143 } 

144 return signature 7 ctx1aghiqfr

145 

146 def _run(self): (empty)

147 

148 out_files = [] 3 ctx1qfr

149 

150 # Detect and upload the wiring file 

151 wiring_file = next(self.session_path.joinpath(self.device_collection, self.pname).glob('*wiring.json'), None) 3 ctx1qfr

152 if wiring_file is not None: 3 ctx1qfr

153 out_files.append(wiring_file) 3 ctx1qfr

154 

155 ephys_files = spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname)) 3 ctx1qfr

156 ephys_files += spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname), ext="ch") 3 ctx1qfr

157 ephys_files += spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname), ext="meta") 3 ctx1qfr

158 

159 for ef in ephys_files: 3 ctx1qfr

160 for typ in ["ap", "lf"]: 3 ctx1qfr

161 bin_file = ef.get(typ) 3 ctx1qfr

162 if not bin_file: 3 ctx1qfr

163 continue 

164 if bin_file.suffix.find("bin") == 1: 3 ctx1qfr

165 with spikeglx.Reader(bin_file) as sr: 1 ctx1f

166 if sr.is_mtscomp: 1 ctx1f

167 out_files.append(bin_file) 

168 else: 

169 _logger.info(f"Compressing binary file {bin_file}") 1 ctx1f

170 cbin_file = sr.compress_file() 1 ctx1f

171 sr.close() 1 ctx1f

172 bin_file.unlink() 1 ctx1f

173 out_files.append(cbin_file) 1 ctx1f

174 out_files.append(cbin_file.with_suffix('.ch')) 1 ctx1f

175 else: 

176 out_files.append(bin_file) 3 ctx1qfr

177 

178 return out_files 3 ctx1qfr

179 

180 

181class EphysCompressNP21(base_tasks.EphysTask): (empty)

182 priority = 90 (empty)

183 cpu = 2 (empty)

184 io_charge = 100 # this jobs reads raw ap files (empty)

185 job_size = 'large' (empty)

186 

187 @property (empty)

188 def signature(self): (empty)

189 signature = { 3 ctx1lmn

190 'input_files': [('*ap.meta', f'{self.device_collection}/{self.pname}', True), 

191 ('*ap.*bin', f'{self.device_collection}/{self.pname}', True), 

192 ('*wiring.json', f'{self.device_collection}/{self.pname}', False)], 

193 'output_files': [('*ap.meta', f'{self.device_collection}/{self.pname}', True), 

194 ('*ap.cbin', f'{self.device_collection}/{self.pname}', True), 

195 ('*ap.ch', f'{self.device_collection}/{self.pname}', True), 

196 ('*lf.meta', f'{self.device_collection}/{self.pname}', True), 

197 ('*lf.cbin', f'{self.device_collection}/{self.pname}', True), 

198 ('*lf.ch', f'{self.device_collection}/{self.pname}', True), 

199 ('*wiring.json', f'{self.device_collection}/{self.pname}', False)] 

200 } 

201 return signature 3 ctx1lmn

202 

203 def _run(self): (empty)

204 

205 out_files = [] 3 ctx1lmn

206 # Detect wiring files 

207 wiring_file = next(self.session_path.joinpath(self.device_collection, self.pname).glob('*wiring.json'), None) 3 ctx1lmn

208 if wiring_file is not None: 3 ctx1lmn

209 out_files.append(wiring_file) 3 ctx1lmn

210 

211 ephys_files = spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname)) 3 ctx1lmn

212 if len(ephys_files) > 0: 3 ctx1lmn

213 bin_file = ephys_files[0].get('ap', None) 2 ctx1lm

214 

215 # This is the case where no ap.bin/.cbin file exists 

216 if len(ephys_files) == 0 or not bin_file: 3 ctx1lmn

217 ephys_files += spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname), ext="ch") 1 ctx1n

218 ephys_files += spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname), ext="meta") 1 ctx1n

219 for ef in ephys_files: 1 ctx1n

220 for typ in ["ap", "lf"]: 1 ctx1n

221 bin_file = ef.get(typ) 1 ctx1n

222 if bin_file: 1 ctx1n

223 out_files.append(bin_file) 1 ctx1n

224 

225 return out_files 1 ctx1n

226 

227 # If the ap.bin / .cbin file does exists instantiate the NP2converter 

228 np_conv = neuropixel.NP2Converter(bin_file, compress=True) 2 ctx1lm

229 np_conv_status = np_conv.process() 2 ctx1lm

230 np_conv_files = np_conv.get_processed_files_NP21() 2 ctx1lm

231 np_conv.sr.close() 2 ctx1lm

232 

233 # Status = 1 - successfully complete 

234 if np_conv_status == 1: # This is the status that it has completed successfully 2 ctx1lm

235 out_files += np_conv_files 2 ctx1lm

236 return out_files 2 ctx1lm

237 # Status = 0 - Already processed 

238 elif np_conv_status == 0: 

239 ephys_files = spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname)) 

240 ephys_files += spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname), ext="ch") 

241 ephys_files += spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname), ext="meta") 

242 for ef in ephys_files: 

243 for typ in ["ap", "lf"]: 

244 bin_file = ef.get(typ) 

245 if bin_file and bin_file.suffix != '.bin': 

246 out_files.append(bin_file) 

247 

248 return out_files 

249 

250 else: 

251 return 

252 

253 

254class EphysCompressNP24(base_tasks.EphysTask): (empty)

255 """ 

256 Compresses NP2.4 data by splitting into N binary files, corresponding to N shanks 

257 :param pname: a probe name string 

258 :param device_collection: the collection containing the probes (usually 'raw_ephys_data') 

259 :param nshanks: number of shanks used (usually 4 but it may be less depending on electrode map), optional 

260 """ 

261 

262 priority = 90 (empty)

263 cpu = 2 (empty)

264 io_charge = 100 # this jobs reads raw ap files (empty)

265 job_size = 'large' (empty)

266 

267 def __init__(self, session_path, *args, pname=None, device_collection='raw_ephys_data', nshanks=None, **kwargs): (empty)

268 assert pname, "pname is a required argument" 4 ctx1decb

269 if nshanks is None: 4 ctx1decb

270 meta_file = next(session_path.joinpath(device_collection, pname).glob('*ap.meta')) 1 ctx1b

271 nshanks = spikeglx._get_nshanks_from_meta(spikeglx.read_meta_data(meta_file)) 1 ctx1b

272 assert nshanks > 1 4 ctx1decb

273 super(EphysCompressNP24, self).__init__( 4 ctx1decb

274 session_path, *args, pname=pname, device_collection=device_collection, nshanks=nshanks, **kwargs) 

275 

276 @property (empty)

277 def signature(self): (empty)

278 

279 signature = { 4 ctx1decb

280 'input_files': [('*ap.meta', f'{self.device_collection}/{self.pname}', True), 

281 ('*ap.*bin', f'{self.device_collection}/{self.pname}', True), 

282 ('*wiring.json', f'{self.device_collection}/{self.pname}', False)], 

283 'output_files': [('*ap.meta', f'{self.device_collection}/{self.pname}{pext}', True) for pext in self.pextra] + 

284 [('*ap.cbin', f'{self.device_collection}/{self.pname}{pext}', True) for pext in self.pextra] + 

285 [('*ap.ch', f'{self.device_collection}/{self.pname}{pext}', True) for pext in self.pextra] + 

286 [('*lf.meta', f'{self.device_collection}/{self.pname}{pext}', True) for pext in self.pextra] + 

287 [('*lf.cbin', f'{self.device_collection}/{self.pname}{pext}', True) for pext in self.pextra] + 

288 [('*lf.ch', f'{self.device_collection}/{self.pname}{pext}', True) for pext in self.pextra] + 

289 [('*wiring.json', f'{self.device_collection}/{self.pname}{pext}', False) for pext in self.pextra] 

290 } 

291 return signature 4 ctx1decb

292 

293 def _run(self, delete_original=True): (empty)

294 

295 # Do we need the ability to register the files once it already been processed and original file deleted? 

296 

297 files = spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname)) 2 ctx1cb

298 assert len(files) == 1 2 ctx1cb

299 bin_file = files[0].get('ap', None) 2 ctx1cb

300 

301 np_conv = neuropixel.NP2Converter(bin_file, post_check=True, compress=True, delete_original=delete_original) 2 ctx1cb

302 np_conv_status = np_conv.process() 2 ctx1cb

303 out_files = np_conv.get_processed_files_NP24() 2 ctx1cb

304 np_conv.sr.close() 2 ctx1cb

305 

306 if np_conv_status == 1: 2 ctx1cb

307 wiring_file = next(self.session_path.joinpath(self.device_collection, self.pname).glob('*wiring.json'), None) 2 ctx1cb

308 if wiring_file is not None: 2 ctx1cb

309 # copy wiring file to each sub probe directory and add to output files 

310 for pext in self.pextra: 1 ctx1c

311 new_wiring_file = self.session_path.joinpath(self.device_collection, f'{self.pname}{pext}', wiring_file.name) 1 ctx1c

312 shutil.copyfile(wiring_file, new_wiring_file) 1 ctx1c

313 out_files.append(new_wiring_file) 1 ctx1c

314 return out_files 2 ctx1cb

315 elif np_conv_status == 0: 1 ctx1c

316 for pext in self.pextra: 1 ctx1c

317 ephys_files = spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, f'{self.pname}{pext}')) 1 ctx1c

318 ephys_files += spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, 1 ctx1c

319 f'{self.pname}{pext}'), ext="ch") 

320 ephys_files += spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, 1 ctx1c

321 f'{self.pname}{pext}'), ext="meta") 

322 for ef in ephys_files: 1 ctx1c

323 for typ in ["ap", "lf"]: 1 ctx1c

324 bin_file = ef.get(typ) 1 ctx1c

325 if bin_file and bin_file.suffix != '.bin': 1 ctx1c

326 out_files.append(bin_file) 1 ctx1c

327 

328 wiring_file = next(self.session_path.joinpath(self.device_collection, 1 ctx1c

329 f'{self.pname}{pext}').glob('*wiring.json'), None) 

330 if wiring_file is None: 1 ctx1c

331 # See if we have the original wiring file 

332 orig_wiring_file = next(self.session_path.joinpath(self.device_collection, 

333 self.pname).glob('*wiring.json'), None) 

334 if orig_wiring_file is not None: 

335 # copy wiring file to sub probe directory and add to output files 

336 new_wiring_file = self.session_path.joinpath(self.device_collection, f'{self.pname}{pext}', 

337 orig_wiring_file.name) 

338 shutil.copyfile(orig_wiring_file, new_wiring_file) 

339 out_files.append(new_wiring_file) 

340 else: 

341 out_files.append(wiring_file) 1 ctx1c

342 

343 return out_files 1 ctx1c

344 else: 

345 return 

346 

347 

348class EphysSyncPulses(SyncPulses): (empty)

349 

350 priority = 90 (empty)

351 cpu = 2 (empty)

352 io_charge = 30 # this jobs reads raw ap files (empty)

353 job_size = 'small' (empty)

354 

355 @property (empty)

356 def signature(self): (empty)

357 signature = { 7 ctx1aghidev

358 'input_files': [('*nidq.cbin', self.sync_collection, True), 

359 ('*nidq.ch', self.sync_collection, False), 

360 ('*nidq.meta', self.sync_collection, True), 

361 ('*nidq.wiring.json', self.sync_collection, True)], 

362 'output_files': [('_spikeglx_sync.times.npy', self.sync_collection, True), 

363 ('_spikeglx_sync.polarities.npy', self.sync_collection, True), 

364 ('_spikeglx_sync.channels.npy', self.sync_collection, True)] 

365 } 

366 

367 return signature 7 ctx1aghidev

368 

369 

370class EphysPulses(base_tasks.EphysTask): (empty)

371 """ 

372 Extract Pulses from raw electrophysiology data into numpy arrays 

373 Perform the probes synchronisation with nidq (3B) or main probe (3A) 

374 First the job extract the sync pulses from the synchronisation task in all probes, and then perform the 

375 synchronisation with the nidq 

376 

377 :param pname: a list of probes names or a single probe name string 

378 :param device_collection: the collection containing the probes (usually 'raw_ephys_data') 

379 :param sync_collection: the collection containing the synchronisation device - nidq (usually 'raw_ephys_data') 

380 """ 

381 priority = 90 (empty)

382 cpu = 2 (empty)

383 io_charge = 30 # this jobs reads raw ap files (empty)

384 job_size = 'small' (empty)

385 

386 def __init__(self, *args, **kwargs): (empty)

387 super(EphysPulses, self).__init__(*args, **kwargs) 9 ctx1aghidebop

388 assert self.device_collection, "device_collection is a required argument" 9 ctx1aghidebop

389 assert self.sync_collection, "sync_collection is a required argument" 9 ctx1aghidebop

390 self.pname = [self.pname] if isinstance(self.pname, str) else self.pname 9 ctx1aghidebop

391 assert isinstance(self.pname, list), 'pname task argument should be a list or a string' 9 ctx1aghidebop

392 

393 @property (empty)

394 def signature(self): (empty)

395 signature = { 9 ctx1aghidebop

396 'input_files': [('*ap.meta', f'{self.device_collection}/{pname}', True) for pname in self.pname] + 

397 [('*ap.cbin', f'{self.device_collection}/{pname}', True) for pname in self.pname] + 

398 [('*ap.ch', f'{self.device_collection}/{pname}', True) for pname in self.pname] + 

399 [('*ap.wiring.json', f'{self.device_collection}/{pname}', False) for pname in self.pname] + 

400 [('_spikeglx_sync.times.npy', self.sync_collection, True), 

401 ('_spikeglx_sync.polarities.npy', self.sync_collection, True), 

402 ('_spikeglx_sync.channels.npy', self.sync_collection, True)], 

403 'output_files': [(f'_spikeglx_sync.times.{pname}.npy', f'{self.device_collection}/{pname}', True) 

404 for pname in self.pname] + 

405 [(f'_spikeglx_sync.polarities.{pname}.npy', f'{self.device_collection}/{pname}', True) 

406 for pname in self.pname] + 

407 [(f'_spikeglx_sync.channels.{pname}.npy', f'{self.device_collection}/{pname}', True) 

408 for pname in self.pname] + 

409 [('*sync.npy', f'{self.device_collection}/{pname}', True) for pname in 

410 self.pname] + 

411 [('*timestamps.npy', f'{self.device_collection}/{pname}', True) for pname in 

412 self.pname] 

413 } 

414 

415 return signature 9 ctx1aghidebop

416 

417 def _run(self, overwrite=False): (empty)

418 

419 out_files = [] 3 ctx1bop

420 for probe in self.pname: 3 ctx1bop

421 files = spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, probe)) 3 ctx1bop

422 assert len(files) == 1 # will error if the session is split 3 ctx1bop

423 bin_file = files[0].get('ap', None) 3 ctx1bop

424 if not bin_file: 3 ctx1bop

425 return [] 

426 _, out = extract_sync(self.session_path, ephys_files=files, overwrite=overwrite) 3 ctx1bop

427 out_files += out 3 ctx1bop

428 

429 status, sync_files = sync_probes.sync(self.session_path, probe_names=self.pname) 3 ctx1bop

430 

431 return out_files + sync_files 3 ctx1bop

432 

433 

434class RawEphysQC(base_tasks.EphysTask): (empty)

435 

436 cpu = 2 (empty)

437 io_charge = 30 # this jobs reads raw ap files (empty)

438 priority = 10 # a lot of jobs depend on this one (empty)

439 job_size = 'small' (empty)

440 

441 @property (empty)

442 def signature(self): (empty)

443 signature = { 6 ctx1aghide

444 'input_files': [('*ap.meta', f'{self.device_collection}/{self.pname}', True), 

445 ('*lf.meta', f'{self.device_collection}/{self.pname}', True), 

446 ('*lf.ch', f'{self.device_collection}/{self.pname}', False), 

447 ('*lf.*bin', f'{self.device_collection}/{self.pname}', False)], 

448 'output_files': [('_iblqc_ephysChannels.apRMS.npy', f'{self.device_collection}/{self.pname}', True), 

449 ('_iblqc_ephysChannels.rawSpikeRates.npy', f'{self.device_collection}/{self.pname}', True), 

450 ('_iblqc_ephysChannels.labels.npy', f'{self.device_collection}/{self.pname}', True), 

451 ('_iblqc_ephysSpectralDensityLF.freqs.npy', f'{self.device_collection}/{self.pname}', True), 

452 ('_iblqc_ephysSpectralDensityLF.power.npy', f'{self.device_collection}/{self.pname}', True), 

453 ('_iblqc_ephysSpectralDensityAP.freqs.npy', f'{self.device_collection}/{self.pname}', True), 

454 ('_iblqc_ephysSpectralDensityAP.power.npy', f'{self.device_collection}/{self.pname}', True), 

455 ('_iblqc_ephysTimeRmsLF.rms.npy', f'{self.device_collection}/{self.pname}', True), 

456 ('_iblqc_ephysTimeRmsLF.timestamps.npy', f'{self.device_collection}/{self.pname}', True)] 

457 } 

458 return signature 6 ctx1aghide

459 

460 # TODO make sure this works with NP2 probes (at the moment not sure it will due to raiseError mapping) 

461 def _run(self, overwrite=False): (empty)

462 

463 eid = self.one.path2eid(self.session_path) 

464 probe = self.one.alyx.rest('insertions', 'list', session=eid, name=self.pname) 

465 

466 # We expect there to only be one probe 

467 if len(probe) != 1: 

468 _logger.warning(f"{self.pname} for {eid} not found") # Should we create it? 

469 self.status = -1 

470 return 

471 

472 pid = probe[0]['id'] 

473 qc_files = [] 

474 _logger.info(f"\nRunning QC for probe insertion {self.pname}") 

475 try: 

476 eqc = ephysqc.EphysQC(pid, session_path=self.session_path, one=self.one) 

477 qc_files.extend(eqc.run(update=True, overwrite=overwrite)) 

478 _logger.info("Creating LFP QC plots") 

479 plot_task = LfpPlots(pid, session_path=self.session_path, one=self.one) 

480 _ = plot_task.run() 

481 self.plot_tasks.append(plot_task) 

482 plot_task = BadChannelsAp(pid, session_path=self.session_path, one=self.one) 

483 _ = plot_task.run() 

484 self.plot_tasks.append(plot_task) 

485 

486 except AssertionError: 

487 _logger.error(traceback.format_exc()) 

488 self.status = -1 

489 

490 return qc_files 

491 

492 

493class CellQCMixin: (empty)

494 """ 

495 This mixin class is used to compute the cell QC metrics and update the json field of the probe insertion 

496 The compute_cell_qc method is static and can be used independently. 

497 """ 

498 @staticmethod (empty)

499 def compute_cell_qc(folder_alf_probe): (empty)

500 """ 

501 Computes the cell QC given an extracted probe alf path 

502 :param folder_alf_probe: folder 

503 :return: 

504 """ 

505 # compute the straight qc 

506 _logger.info(f"Computing cluster qc for {folder_alf_probe}") 

507 spikes = alfio.load_object(folder_alf_probe, 'spikes') 

508 clusters = alfio.load_object(folder_alf_probe, 'clusters') 

509 df_units, drift = ephysqc.spike_sorting_metrics( 

510 spikes.times, spikes.clusters, spikes.amps, spikes.depths, 

511 cluster_ids=np.arange(clusters.channels.size)) 

512 # if the ks2 labels file exist, load them and add the column 

513 file_labels = folder_alf_probe.joinpath('cluster_KSLabel.tsv') 

514 if file_labels.exists(): 

515 ks2_labels = pd.read_csv(file_labels, sep='\t') 

516 ks2_labels.rename(columns={'KSLabel': 'ks2_label'}, inplace=True) 

517 df_units = pd.concat( 

518 [df_units, ks2_labels['ks2_label'].reindex(df_units.index)], axis=1) 

519 # save as parquet file 

520 df_units.to_parquet(folder_alf_probe.joinpath("clusters.metrics.pqt")) 

521 return folder_alf_probe.joinpath("clusters.metrics.pqt"), df_units, drift 

522 

523 def _label_probe_qc(self, folder_probe, df_units, drift): (empty)

524 """ 

525 Labels the json field of the alyx corresponding probe insertion 

526 :param folder_probe: 

527 :param df_units: 

528 :param drift: 

529 :return: 

530 """ 

531 eid = self.one.path2eid(self.session_path, query_type='remote') 

532 pdict = self.one.alyx.rest('insertions', 'list', session=eid, name=self.pname, no_cache=True) 

533 if len(pdict) != 1: 

534 _logger.warning(f'No probe found for probe name: {self.pname}') 

535 return 

536 isok = df_units['label'] == 1 

537 qcdict = {'n_units': int(df_units.shape[0]), 

538 'n_units_qc_pass': int(np.sum(isok)), 

539 'firing_rate_max': np.max(df_units['firing_rate'][isok]), 

540 'firing_rate_median': np.median(df_units['firing_rate'][isok]), 

541 'amplitude_max_uV': np.max(df_units['amp_max'][isok]) * 1e6, 

542 'amplitude_median_uV': np.max(df_units['amp_median'][isok]) * 1e6, 

543 'drift_rms_um': rms(drift['drift_um']), 

544 } 

545 file_wm = folder_probe.joinpath('_kilosort_whitening.matrix.npy') 

546 if file_wm.exists(): 

547 wm = np.load(file_wm) 

548 qcdict['whitening_matrix_conditioning'] = np.linalg.cond(wm) 

549 # groom qc dict (this function will eventually go directly into the json field update) 

550 for k in qcdict: 

551 if isinstance(qcdict[k], np.int64): 

552 qcdict[k] = int(qcdict[k]) 

553 elif isinstance(qcdict[k], float): 

554 qcdict[k] = np.round(qcdict[k], 2) 

555 self.one.alyx.json_field_update("insertions", pdict[0]["id"], "json", qcdict) 

556 

557 

558class SpikeSorting(base_tasks.EphysTask, CellQCMixin): (empty)

559 """ 

560 Pykilosort 2.5 pipeline 

561 """ 

562 gpu = 1 (empty)

563 io_charge = 100 # this jobs reads raw ap files (empty)

564 priority = 60 (empty)

565 job_size = 'large' (empty)

566 force = True (empty)

567 

568 SHELL_SCRIPT = Path.home().joinpath( (empty)

569 "Documents/PYTHON/iblscripts/deploy/serverpc/iblsorter/run_iblsorter.sh" 

570 ) 

571 SPIKE_SORTER_NAME = 'iblsorter' (empty)

572 PYKILOSORT_REPO = Path.home().joinpath('Documents/PYTHON/SPIKE_SORTING/ibl-sorter') (empty)

573 

574 @property (empty)

575 def signature(self): (empty)

576 signature = { 6 ctx1aghide

577 'input_files': [('*ap.meta', f'{self.device_collection}/{self.pname}', True), 

578 ('*ap.*bin', f'{self.device_collection}/{self.pname}', True), 

579 ('*ap.ch', f'{self.device_collection}/{self.pname}', False), 

580 ('*sync.npy', f'{self.device_collection}/{self.pname}', True)], 

581 'output_files': [('spike_sorting_pykilosort.log', f'spike_sorters/pykilosort/{self.pname}', True), 

582 ('_iblqc_ephysTimeRmsAP.rms.npy', f'{self.device_collection}/{self.pname}', True), 

583 ('_iblqc_ephysTimeRmsAP.timestamps.npy', f'{self.device_collection}/{self.pname}', True)] 

584 } 

585 return signature 6 ctx1aghide

586 

587 @staticmethod (empty)

588 def _sample2v(ap_file): (empty)

589 md = spikeglx.read_meta_data(ap_file.with_suffix(".meta")) 

590 s2v = spikeglx._conversion_sample2v_from_meta(md) 

591 return s2v["ap"][0] 

592 

593 @staticmethod (empty)

594 def _fetch_iblsorter_version(repo_path): (empty)

595 try: 

596 import iblsorter 

597 return f"iblsorter_{iblsorter.__version__}" 

598 except ImportError: 

599 _logger.info('IBL-sorter not in environment, trying to locate the repository') 

600 init_file = Path(repo_path).joinpath('ibl-sorter', '__init__.py') 

601 try: 

602 with open(init_file) as fid: 

603 lines = fid.readlines() 

604 for line in lines: 

605 if line.startswith("__version__ = "): 

606 version = line.split('=')[-1].strip().replace('"', '').replace("'", '') 

607 except Exception: 

608 pass 

609 return f"iblsorter_{version}" 

610 

611 @staticmethod (empty)

612 def _fetch_iblsorter_run_version(log_file): (empty)

613 """ 

614 Parse the following line (2 formats depending on version) from the log files to get the version 

615 '\x1b[0m15:39:37.919 [I] ibl:90 Starting Pykilosort version ibl_1.2.1, output in gnagga^[[0m\n' 

616 '\x1b[0m15:39:37.919 [I] ibl:90 Starting Pykilosort version ibl_1.3.0^[[0m\n' 

617 """ 

618 with open(log_file) as fid: 1 ctx1t

619 line = fid.readline() 1 ctx1t

620 version = re.search('version (.*), output', line) 1 ctx1t

621 version = version or re.search('version (.*)', line) # old versions have output, new have a version line 1 ctx1t

622 version = re.sub(r'\^\[{2}[0-9]+m', '', version.group(1)) # removes the coloring tags 1 ctx1t

623 return version 1 ctx1t

624 

625 def _run_iblsort(self, ap_file): (empty)

626 """ 

627 Runs the ks2 matlab spike sorting for one probe dataset 

628 the raw spike sorting output is in session_path/spike_sorters/{self.SPIKE_SORTER_NAME}/probeXX folder 

629 (discontinued support for old spike sortings in the probe folder <1.5.5) 

630 :return: path of the folder containing ks2 spike sorting output 

631 """ 

632 self.version = self._fetch_iblsorter_version(self.PYKILOSORT_REPO) 

633 label = ap_file.parts[-2] # this is usually the probe name 

634 sorter_dir = self.session_path.joinpath("spike_sorters", self.SPIKE_SORTER_NAME, label) 

635 self.FORCE_RERUN = False 

636 if not self.FORCE_RERUN: 

637 log_file = sorter_dir.joinpath(f"spike_sorting_{self.SPIKE_SORTER_NAME}.log") 

638 if log_file.exists(): 

639 run_version = self._fetch_iblsorter_run_version(log_file) 

640 if packaging.version.parse(run_version) >= packaging.version.parse('1.7.0'): 

641 _logger.info(f"Already ran: spike_sorting_{self.SPIKE_SORTER_NAME}.log" 

642 f" found in {sorter_dir}, skipping.") 

643 return sorter_dir 

644 else: 

645 self.FORCE_RERUN = True 

646 # get the scratch drive from the shell script 

647 with open(self.SHELL_SCRIPT) as fid: 

648 lines = fid.readlines() 

649 line = [line for line in lines if line.startswith("SCRATCH_DRIVE=")][0] 

650 m = re.search(r"\=(.*?)(\#|\n)", line)[0] 

651 scratch_drive = Path(m[1:-1].strip()) 

652 assert scratch_drive.exists() 

653 spikesorter_dir = f"{self.version}_{'_'.join(list(self.session_path.parts[-3:]))}_{self.pname}" 

654 temp_dir = scratch_drive.joinpath(spikesorter_dir) 

655 _logger.info(f"job progress command: tail -f {temp_dir} *.log") 

656 temp_dir.mkdir(parents=True, exist_ok=True) 

657 check_nvidia_driver() 

658 try: 

659 # if pykilosort is in the environment, use the installed version within the task 

660 import iblsorter.ibl # noqa 

661 iblsorter.ibl.run_spike_sorting_ibl(bin_file=ap_file, scratch_dir=temp_dir) 

662 except ImportError: 

663 command2run = f"{self.SHELL_SCRIPT} {ap_file} {temp_dir}" 

664 _logger.info(command2run) 

665 process = subprocess.Popen( 

666 command2run, 

667 shell=True, 

668 stdout=subprocess.PIPE, 

669 stderr=subprocess.PIPE, 

670 executable="/bin/bash", 

671 ) 

672 info, error = process.communicate()