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

1import logging (empty)

2from pathlib import Path (empty)

3import re (empty)

4import shutil (empty)

5import subprocess (empty)

6import traceback (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)

16import iblutil.util (empty)

17 

18from ibllib.misc import check_nvidia_driver (empty)

19from ibllib.pipes import base_tasks (empty)

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

21from ibllib.ephys import ephysqc (empty)

22import ibllib.ephys.spikes (empty)

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

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

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

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

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

28 

29 

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

31 

32 

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

34 """ 

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

36 """ 

37 

38 priority = 100 (empty)

39 job_size = 'small' (empty)

40 

41 @property (empty)

42 def signature(self): (empty)

43 signature = { 1 ctx1u

44 'input_files': [], 

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

46 } 

47 return signature 1 ctx1u

48 

49 def _run(self): (empty)

50 

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

52 

53 return out_files 1 ctx1u

54 

55 

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

57 """ 

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

59 """ 

60 

61 priority = 90 (empty)

62 cpu = 2 (empty)

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

64 job_size = 'small' (empty)

65 

66 @property (empty)

67 def signature(self): (empty)

68 signature = { 9 ctx1arstopefn

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

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

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

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

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

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

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

76 } 

77 return signature 9 ctx1arstopefn

78 

79 def _run(self): (empty)

80 

81 out_files = [] 3 ctx1efn

82 

83 # Detect the wiring file 

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

85 if wiring_file is not None: 3 ctx1efn

86 out_files.append(wiring_file) 3 ctx1efn

87 

88 # Search for .bin files in the sync_collection folder 

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

90 bin_file = files[0] if len(files) == 1 else None 3 ctx1efn

91 

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

93 if bin_file is None: 3 ctx1efn

94 for ext in ['ch', 'meta']: 1 ctx1n

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

96 ext_file = files[0] if len(files) == 1 else None 1 ctx1n

97 if ext_file is not None: 1 ctx1n

98 out_files.append(ext_file) 1 ctx1n

99 

100 return out_files if len(out_files) > 0 else None 1 ctx1n

101 

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

103 sr = spikeglx.Reader(bin_file) 2 ctx1ef

104 if sr.is_mtscomp: 2 ctx1ef

105 sr.close() 1 ctx1e

106 cbin_file = bin_file 1 ctx1e

107 assert cbin_file.suffix == '.cbin' 1 ctx1e

108 else: 

109 cbin_file = sr.compress_file() 1 ctx1f

110 sr.close() 1 ctx1f

111 bin_file.unlink() 1 ctx1f

112 

113 meta_file = cbin_file.with_suffix('.meta') 2 ctx1ef

114 ch_file = cbin_file.with_suffix('.ch') 2 ctx1ef

115 

116 out_files.append(cbin_file) 2 ctx1ef

117 out_files.append(ch_file) 2 ctx1ef

118 out_files.append(meta_file) 2 ctx1ef

119 

120 return out_files 2 ctx1ef

121 

122 

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

124 priority = 90 (empty)

125 cpu = 2 (empty)

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

127 job_size = 'small' (empty)

128 

129 @property (empty)

130 def signature(self): (empty)

131 signature = { 3 ctx1ldm

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

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

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

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

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

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

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

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

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

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

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

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

144 } 

145 return signature 3 ctx1ldm

146 

147 def _run(self): (empty)

148 

149 out_files = [] 3 ctx1ldm

150 

151 # Detect and upload the wiring file 

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

153 if wiring_file is not None: 3 ctx1ldm

154 out_files.append(wiring_file) 3 ctx1ldm

155 

156 ephys_files = spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname)) 3 ctx1ldm

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

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

159 

160 for ef in ephys_files: 3 ctx1ldm

161 for typ in ["ap", "lf"]: 3 ctx1ldm

162 bin_file = ef.get(typ) 3 ctx1ldm

163 if not bin_file: 3 ctx1ldm

164 continue 

165 if bin_file.suffix.find("bin") == 1: 3 ctx1ldm

166 with spikeglx.Reader(bin_file) as sr: 1 ctx1d

167 if sr.is_mtscomp: 1 ctx1d

168 out_files.append(bin_file) 

169 else: 

170 _logger.info(f"Compressing binary file {bin_file}") 1 ctx1d

171 cbin_file = sr.compress_file() 1 ctx1d

172 sr.close() 1 ctx1d

173 bin_file.unlink() 1 ctx1d

174 out_files.append(cbin_file) 1 ctx1d

175 out_files.append(cbin_file.with_suffix('.ch')) 1 ctx1d

176 else: 

177 out_files.append(bin_file) 3 ctx1ldm

178 

179 return out_files 3 ctx1ldm

180 

181 

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

183 priority = 90 (empty)

184 cpu = 2 (empty)

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

186 job_size = 'large' (empty)

187 

188 @property (empty)

189 def signature(self): (empty)

190 signature = { 3 ctx1ghi

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

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

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

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

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

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

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

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

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

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

201 } 

202 return signature 3 ctx1ghi

203 

204 def _run(self): (empty)

205 

206 out_files = [] 3 ctx1ghi

207 # Detect wiring files 

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

209 if wiring_file is not None: 3 ctx1ghi

210 out_files.append(wiring_file) 3 ctx1ghi

211 

212 ephys_files = spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, self.pname)) 3 ctx1ghi

213 if len(ephys_files) > 0: 3 ctx1ghi

214 bin_file = ephys_files[0].get('ap', None) 2 ctx1gh

215 

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

217 if len(ephys_files) == 0 or not bin_file: 3 ctx1ghi

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

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

220 for ef in ephys_files: 1 ctx1i

221 for typ in ["ap", "lf"]: 1 ctx1i

222 bin_file = ef.get(typ) 1 ctx1i

223 if bin_file: 1 ctx1i

224 out_files.append(bin_file) 1 ctx1i

225 

226 return out_files 1 ctx1i

227 

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

229 np_conv = neuropixel.NP2Converter(bin_file, compress=True) 2 ctx1gh

230 np_conv_status = np_conv.process() 2 ctx1gh

231 np_conv_files = np_conv.get_processed_files_NP21() 2 ctx1gh

232 np_conv.sr.close() 2 ctx1gh

233 

234 # Status = 1 - successfully complete 

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

236 out_files += np_conv_files 2 ctx1gh

237 return out_files 2 ctx1gh

238 # Status = 0 - Already processed 

239 elif np_conv_status == 0: 

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

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

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

243 for ef in ephys_files: 

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

245 bin_file = ef.get(typ) 

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

247 out_files.append(bin_file) 

248 

249 return out_files 

250 

251 else: 

252 return 

253 

254 

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

256 """ 

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

258 :param pname: a probe name string 

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

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

261 """ 

262 

263 priority = 90 (empty)

264 cpu = 2 (empty)

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

266 job_size = 'large' (empty)

267 

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

269 assert pname, "pname is a required argument" 4 ctx1opcb

270 if nshanks is None: 4 ctx1opcb

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

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

273 assert nshanks > 1 4 ctx1opcb

274 super(EphysCompressNP24, self).__init__( 4 ctx1opcb

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

276 

277 @property (empty)

278 def signature(self): (empty)

279 

280 signature = { 2 ctx1cb

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

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

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

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

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

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

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

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

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

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

291 } 

292 return signature 2 ctx1cb

293 

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

295 

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

297 

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

299 assert len(files) == 1 2 ctx1cb

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

301 

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

303 np_conv_status = np_conv.process() 2 ctx1cb

304 out_files = np_conv.get_processed_files_NP24() 2 ctx1cb

305 np_conv.sr.close() 2 ctx1cb

306 

307 if np_conv_status == 1: 2 ctx1cb

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

309 if wiring_file is not None: 2 ctx1cb

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

311 for pext in self.pextra: 1 ctx1c

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

313 shutil.copyfile(wiring_file, new_wiring_file) 1 ctx1c

314 out_files.append(new_wiring_file) 1 ctx1c

315 return out_files 2 ctx1cb

316 elif np_conv_status == 0: 1 ctx1c

317 for pext in self.pextra: 1 ctx1c

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

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

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

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

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

323 for ef in ephys_files: 1 ctx1c

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

325 bin_file = ef.get(typ) 1 ctx1c

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

327 out_files.append(bin_file) 1 ctx1c

328 

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

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

331 if wiring_file is None: 1 ctx1c

332 # See if we have the original wiring file 

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

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

335 if orig_wiring_file is not None: 

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

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

338 orig_wiring_file.name) 

339 shutil.copyfile(orig_wiring_file, new_wiring_file) 

340 out_files.append(new_wiring_file) 

341 else: 

342 out_files.append(wiring_file) 1 ctx1c

343 

344 return out_files 1 ctx1c

345 else: 

346 return 

347 

348 

349class EphysSyncPulses(SyncPulses): (empty)

350 

351 priority = 90 (empty)

352 cpu = 2 (empty)

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

354 job_size = 'small' (empty)

355 

356 @property (empty)

357 def signature(self): (empty)

358 signature = { 7 ctx1arstopv

359 'input_files': [('*nidq.cbin', self.sync_collection, False), 

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

361 ('*nidq.meta', self.sync_collection, False), 

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

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

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

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

366 } 

367 

368 return signature 7 ctx1arstopv

369 

370 

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

372 """ 

373 Extract Pulses from raw electrophysiology data into numpy arrays 

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

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

376 synchronisation with the nidq 

377 

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

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

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

381 """ 

382 priority = 90 (empty)

383 cpu = 2 (empty)

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

385 job_size = 'small' (empty)

386 

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

388 super(EphysPulses, self).__init__(*args, **kwargs) 9 ctx1arstopbjk

389 assert self.device_collection, "device_collection is a required argument" 9 ctx1arstopbjk

390 assert self.sync_collection, "sync_collection is a required argument" 9 ctx1arstopbjk

391 self.pname = [self.pname] if isinstance(self.pname, str) else self.pname 9 ctx1arstopbjk

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

393 

394 @property (empty)

395 def signature(self): (empty)

396 signature = { 3 ctx1bjk

397 'input_files': 

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

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

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

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

402 [('_spikeglx_sync.times.*npy', f'{self.device_collection}/{pname}', False) for pname in self.pname] + 

403 [('_spikeglx_sync.polarities.*npy', f'{self.device_collection}/{pname}', False) for pname in self.pname] + 

404 [('_spikeglx_sync.channels.*npy', f'{self.device_collection}/{pname}', False) for pname in self.pname] + 

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

406 ('_spikeglx_sync.polarities.*npy', self.sync_collection, True), 

407 ('_spikeglx_sync.channels.*npy', self.sync_collection, True), 

408 ('*ap.meta', self.sync_collection, True) 

409 ], 

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

411 for pname in self.pname] + 

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

413 for pname in self.pname] + 

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

415 for pname in self.pname] + 

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

417 self.pname] + 

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

419 self.pname] 

420 } 

421 

422 return signature 3 ctx1bjk

423 

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

425 

426 out_files = [] 3 ctx1bjk

427 for probe in self.pname: 3 ctx1bjk

428 files = spikeglx.glob_ephys_files(self.session_path.joinpath(self.device_collection, probe)) 3 ctx1bjk

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

430 bin_file = files[0].get('ap', None) 3 ctx1bjk

431 if not bin_file: 3 ctx1bjk

432 return [] 

433 _, out = extract_sync(self.session_path, ephys_files=files, overwrite=overwrite) 3 ctx1bjk

434 out_files += out 3 ctx1bjk

435 

436 status, sync_files = sync_probes.sync(self.session_path, probe_names=self.pname) 3 ctx1bjk

437 

438 return out_files + sync_files 3 ctx1bjk

439 

440 

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

442 

443 cpu = 2 (empty)

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

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

446 job_size = 'small' (empty)

447 

448 @property (empty)

449 def signature(self): (empty)

450 signature = { 

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

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

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

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

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

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

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

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

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

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

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

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

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

464 } 

465 return signature 

466 

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

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

469 

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

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

472 

473 # We expect there to only be one probe 

474 if len(probe) != 1: 

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

476 self.status = -1 

477 return 

478 

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

480 qc_files = [] 

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

482 try: 

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

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

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

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

487 _ = plot_task.run() 

488 self.plot_tasks.append(plot_task) 

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

490 _ = plot_task.run() 

491 self.plot_tasks.append(plot_task) 

492 

493 except AssertionError: 

494 _logger.error(traceback.format_exc()) 

495 self.status = -1 

496 

497 return qc_files 

498 

499 

500class CellQCMixin: (empty)

501 """ 

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

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

504 """ 

505 @staticmethod (empty)

506 def compute_cell_qc(folder_alf_probe): (empty)

507 """ 

508 Computes the cell QC given an extracted probe alf path 

509 :param folder_alf_probe: folder 

510 :return: 

511 """ 

512 # compute the straight qc 

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

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

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

516 df_units, drift = ephysqc.spike_sorting_metrics( 

517 spikes.times, spikes.clusters, spikes.amps, spikes.depths, 

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

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

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

521 if file_labels.exists(): 

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

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

524 df_units = pd.concat( 

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

526 # save as parquet file 

527 df_units.to_parquet(file_metrics := folder_alf_probe.joinpath("clusters.metrics.pqt")) 

528 

529 assert np.all((df_units['bitwise_fail'] == 0) == (df_units['label'] == 1)) # useless but sanity check for OW 

530 

531 cok = df_units['bitwise_fail'] == 0 

532 sok = cok[spikes['clusters']].values 

533 spikes['templates'] = spikes['templates'].astype(np.uint16) 

534 spikes['clusters'] = spikes['clusters'].astype(np.uint16) 

535 spikes['depths'] = spikes['depths'].astype(np.float32) 

536 spikes['amps'] = spikes['amps'].astype(np.float32) 

537 file_passing = folder_alf_probe.joinpath('passingSpikes.table.pqt') 

538 df_spikes = pd.DataFrame(spikes) 

539 df_spikes = df_spikes.iloc[sok, :].reset_index(drop=True) 

540 df_spikes.to_parquet(file_passing) 

541 

542 return [file_metrics, file_passing], df_units, drift 

543 

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

545 """ 

546 Labels the json field of the alyx corresponding probe insertion 

547 :param folder_probe: 

548 :param df_units: 

549 :param drift: 

550 :return: 

551 """ 

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

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

554 if len(pdict) != 1: 

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

556 return 

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

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

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

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

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

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

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

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

565 } 

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

567 if file_wm.exists(): 

568 wm = np.load(file_wm) 

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

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

571 for k in qcdict: 

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

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

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

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

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

577 

578 

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

580 """ 

581 Pykilosort 2.5 pipeline 

582 """ 

583 gpu = 1 (empty)

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

585 priority = 60 (empty)

586 job_size = 'large' (empty)

587 force = True (empty)

588 env = 'iblsorter' (empty)

589 _sortername = 'iblsorter' (empty)

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

591 f"Documents/PYTHON/iblscripts/deploy/serverpc/{_sortername}/sort_recording.sh" 

592 ) 

593 SPIKE_SORTER_NAME = 'iblsorter' (empty)

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

595 

596 @property (empty)

597 def signature(self): (empty)

598 signature = { 

599 'input_files': [ 

600 ('*ap.meta', f'{self.device_collection}/{self.pname}', True), 

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

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

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

604 ], 

605 'output_files': [ 

606 # ./raw_ephys_data/{self.pname}/ 

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

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

609 ('_iblqc_ephysSaturation.samples.npy', f'{self.device_collection}/{self.pname}/', True), 

610 # ./spike_sorters/iblsorter/{self.pname} 

611 ('_kilosort_raw.output.tar', f'spike_sorters/{self._sortername}/{self.pname}/', True), 

612 # ./alf/{self.pname}/iblsorter 

613 (f'_ibl_log.info_{self.SPIKE_SORTER_NAME}.log', f'alf/{self.pname}/{self._sortername}', True), 

614 ('_kilosort_whitening.matrix.npy', f'alf/{self.pname}/{self._sortername}/', True), 

615 ('_phy_spikes_subset.channels.npy', f'alf/{self.pname}/{self._sortername}/', True), 

616 ('_phy_spikes_subset.spikes.npy', f'alf/{self.pname}/{self._sortername}/', True), 

617 ('_phy_spikes_subset.waveforms.npy', f'alf/{self.pname}/{self._sortername}/', True), 

618 ('channels.labels.npy', f'alf/{self.pname}/{self._sortername}/', True), 

619 ('channels.localCoordinates.npy', f'alf/{self.pname}/{self._sortername}/', True), 

620 ('channels.rawInd.npy', f'alf/{self.pname}/{self._sortername}/', True), 

621 ('clusters.amps.npy', f'alf/{self.pname}/{self._sortername}/', True), 

622 ('clusters.channels.npy', f'alf/{self.pname}/{self._sortername}/', True), 

623 ('clusters.depths.npy', f'alf/{self.pname}/{self._sortername}/', True), 

624 ('clusters.metrics.pqt', f'alf/{self.pname}/{self._sortername}/', True), 

625 ('clusters.peakToTrough.npy', f'alf/{self.pname}/{self._sortername}/', True), 

626 ('clusters.uuids.csv', f'alf/{self.pname}/{self._sortername}/', True), 

627 ('clusters.waveforms.npy', f'alf/{self.pname}/{self._sortername}/', True), 

628 ('clusters.waveformsChannels.npy', f'alf/{self.pname}/{self._sortername}/', True), 

629 ('drift.times.npy', f'alf/{self.pname}/{self._sortername}/', True), 

630 ('drift.um.npy', f'alf/{self.pname}/{self._sortername}/', True), 

631 ('drift_depths.um.npy', f'alf/{self.pname}/{self._sortername}/', True), 

632 ('passingSpikes.table.pqt', f'alf/{self.pname}/{self._sortername}/', True), 

633 ('spikes.amps.npy', f'alf/{self.pname}/{self._sortername}/', True), 

634 ('spikes.clusters.npy', f'alf/{self.pname}/{self._sortername}/', True), 

635 ('spikes.depths.npy', f'alf/{self.pname}/{self._sortername}/', True), 

636 ('spikes.samples.npy', f'alf/{self.pname}/{self._sortername}/', True), 

637 ('spikes.templates.npy', f'alf/{self.pname}/{self._sortername}/', True), 

638 ('spikes.times.npy', f'alf/{self.pname}/{self._sortername}/', True), 

639 ('templates.amps.npy', f'alf/{self.pname}/{self._sortername}/', True), 

640 ('templates.waveforms.npy', f'alf/{self.pname}/{self._sortername}/', True), 

641 ('templates.waveformsChannels.npy', f'alf/{self.pname}/{self._sortername}/', True), 

642 ('waveforms.channels.npz', f'alf/{self.pname}/{self._sortername}/', True), 

643 ('waveforms.table.pqt', f'alf/{self.pname}/{self._sortername}/', True), 

644 ('waveforms.templates.npy', f'alf/{self.pname}/{self._sortername}/', True), 

645 ('waveforms.traces.npy', f'alf/{self.pname}/{self._sortername}/', True), 

646 ], 

647 } 

648 return signature 

649 

650 @property (empty)

651 def scratch_folder_run(self): (empty)

652 """ 

653 Constructs a path to a temporary folder for the spike sorting output and scratch files 

654 This is usually on a high performance drive, and we should factor around 2.5 times the uncompressed raw recording size 

655 For a scratch drive at /mnt/h0 we would have the following temp dir: 

656 /mnt/h0/iblsorter_1.8.0_CSHL071_2020-10-04_001_probe01/ 

657 """ 

658 # get the scratch drive from the shell script 

659 if self.scratch_folder is None: 

660 with open(self.SHELL_SCRIPT) as fid: 

661 lines = fid.readlines() 

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

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

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