_common.py

50 lines · skills/neuropixels-analysis/scripts/_common.py at 5f2bc03

1"""Recording contracts shared by the bundled Neuropixels commands."""

2from pathlib import Path

4import numpy as np

5import spikeinterface.full as si

8def load_saved(path, child):

9 """Accept an extractor folder or a bundled command's output parent."""

10 path = Path(path)

11 return si.load(path / child if (path / child).is_dir() else path)

14def validate_recording(recording):

15 """These commands process one continuous, calibrated probe recording."""

16 if recording.get_num_segments() != 1:

17 raise ValueError("Select one recording segment before using this command; preserve its time origin.")

18 fs = recording.get_sampling_frequency()

19 if not np.isfinite(fs) or fs <= 0 or recording.get_num_samples() == 0:

20 raise ValueError("Recording must have samples and a positive finite sampling frequency.")

21 locations = recording.get_channel_locations()

22 if locations.shape[0] != recording.get_num_channels() or not np.isfinite(locations).all():

23 raise ValueError("Every channel needs a finite location in micrometers.")

24 if np.unique(locations, axis=0).shape[0] != len(locations):

25 raise ValueError("Channel locations must be unique; verify shank geometry and channel mapping.")

26 if not recording.has_scaleable_traces():

27 raise ValueError("Recording needs calibrated gain_to_uV/offset_to_uV before amplitude analysis.")

30def apply_phase_correction(recording):

31 """Use recorded ADC offsets, never infer them from a probe-generation label."""

32 shifts = recording.get_property('inter_sample_shift')

33 if shifts is None or not np.isfinite(shifts).all():

34 raise ValueError("Missing inter_sample_shift metadata; verify ADC timing or explicitly skip phase correction.")

35 return si.phase_shift(recording)

38def reference_by_shank(recording):

39 """Median reference each shank using contact-to-device indices, not channel IDs."""

40 probe = recording.get_probe() # Deliberately requires a single attached probe.

41 indices = np.asarray(probe.device_channel_indices)

42 if sorted(indices.tolist()) != list(range(recording.get_num_channels())):

43 raise ValueError("Probe device-channel mapping must cover the selected channels exactly once.")

44 if probe.shank_ids is None:

45 raise ValueError("Missing probe shank_ids; annotate the verified shank assignment before referencing.")

46 shanks = np.asarray(probe.shank_ids)

47 groups = [recording.channel_ids[indices[shanks == shank]].tolist() for shank in np.unique(shanks)]

48 if any(len(group) < 2 for group in groups):

49 raise ValueError("Median reference needs at least two retained channels on every shank.")

50 return si.common_reference(recording, operator='median', reference='global', groups=groups)