Skip to content

spikesorting_artifact.py

ArtifactDetectionParameters

Bases: SpyglassMixin, Manual

Parameters for detecting artifact times within a sort group.

Parameters:

Name Type Description Default
artifact_params_name str

Name of the artifact detection parameters.

 required
artifact_params dict

Dictionary of parameters for artifact detection, including: zscore_thresh : float or None Stdev threshold for exclusion, should be >=0 amplitude_thresh : float or None Amplitude threshold for exclusion, should be >=0 proportion_above_thresh : float Proportion of electrodes that need to have threshold crossings, should be >0 and <=1 removal_window_ms : float Width of the window in milliseconds to mask out per artifact

 required
Source code in src/spyglass/spikesorting/v0/spikesorting_artifact.py 
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
@schema
class ArtifactDetectionParameters(SpyglassMixin, dj.Manual):
    """Parameters for detecting artifact times within a sort group.

    Parameters
    ----------
    artifact_params_name : str
        Name of the artifact detection parameters.
    artifact_params : dict
        Dictionary of parameters for artifact detection, including:
            zscore_thresh : float or None
                Stdev threshold for exclusion, should be >=0
            amplitude_thresh : float or None
                Amplitude threshold for exclusion, should be >=0
            proportion_above_thresh : float
                Proportion of electrodes that need to have threshold crossings,
                should be >0 and <=1
            removal_window_ms : float
                Width of the window in milliseconds to mask out per artifact
    """

    definition = """
    # Parameters for detecting artifact times within a sort group.
    artifact_params_name: varchar(200)
    ---
    artifact_params: blob  # dictionary of parameters
    """

    def insert_default(self):
        """Insert the default artifact parameters."""
        params_default = {
            "zscore_thresh": 5.0,  # must be None or >= 0
            "amplitude_thresh": 3000,  # must be None or >= 0
            "proportion_above_thresh": 0.5,  # must be > 0 and <= 1
            "removal_window_ms": 1.0,  # in milliseconds
        }
        params_none = {
            "zscore_thresh": None,  # must be None or >= 0
            "amplitude_thresh": None,  # must be None or >= 0
        }
        self.insert(
            [("default", params_default), ("none", params_none)],
            skip_duplicates=True,
        )

insert_default()

Insert the default artifact parameters.

Source code in src/spyglass/spikesorting/v0/spikesorting_artifact.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def insert_default(self):
    """Insert the default artifact parameters."""
    params_default = {
        "zscore_thresh": 5.0,  # must be None or >= 0
        "amplitude_thresh": 3000,  # must be None or >= 0
        "proportion_above_thresh": 0.5,  # must be > 0 and <= 1
        "removal_window_ms": 1.0,  # in milliseconds
    }
    params_none = {
        "zscore_thresh": None,  # must be None or >= 0
        "amplitude_thresh": None,  # must be None or >= 0
    }
    self.insert(
        [("default", params_default), ("none", params_none)],
        skip_duplicates=True,
    )

ArtifactDetection

Bases: SpyglassMixin, Computed

Source code in src/spyglass/spikesorting/v0/spikesorting_artifact.py 
 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
157
@schema
class ArtifactDetection(SpyglassMixin, dj.Computed):
    definition = """
    # Stores artifact times and valid no-artifact times as intervals.
    -> ArtifactDetectionSelection
    ---
    artifact_times: longblob # np array of artifact intervals
    artifact_removed_valid_times: longblob # np array of valid no-artifact intervals
    artifact_removed_interval_list_name: varchar(200) # name of the array of no-artifact valid time intervals
    """

    _parallel_make = True

    def make(self, key):
        """Populate the ArtifactDetection table.

        If custom_artifact_detection is set in selection table, do nothing.

        Fetches...
            - Parameters from ArtifactDetectionParameters
            - Recording from SpikeSortingRecording (loads with spikeinterface)
        Uses module-level function _get_artifact_times to detect artifacts.
        """
        if not (ArtifactDetectionSelection & key).fetch1(
            "custom_artifact_detection"
        ):
            # get the dict of artifact params associated with this artifact_params_name
            artifact_params = (ArtifactDetectionParameters & key).fetch1(
                "artifact_params"
            )

            recording_path = (SpikeSortingRecording & key).fetch1(
                "recording_path"
            )
            recording_name = SpikeSortingRecording._get_recording_name(key)
            recording = si.load_extractor(recording_path)

            job_kwargs = {
                "chunk_duration": "10s",
                "n_jobs": 4,
                "progress_bar": "True",
            }

            artifact_removed_valid_times, artifact_times = _get_artifact_times(
                recording, **artifact_params, **job_kwargs
            )

            key["artifact_times"] = artifact_times
            key["artifact_removed_valid_times"] = artifact_removed_valid_times

            # set up a name for no-artifact times using recording id
            key["artifact_removed_interval_list_name"] = (
                recording_name
                + "_"
                + key["artifact_params_name"]
                + "_artifact_removed_valid_times"
            )

            ArtifactRemovedIntervalList.insert1(key, replace=True)

            # also insert into IntervalList
            tmp_key = {}
            tmp_key["nwb_file_name"] = key["nwb_file_name"]
            tmp_key["interval_list_name"] = key[
                "artifact_removed_interval_list_name"
            ]
            tmp_key["valid_times"] = key["artifact_removed_valid_times"]
            tmp_key["pipeline"] = "spikesorting_artifact_v0"
            IntervalList.insert1(tmp_key, replace=True)

            # insert into computed table
            self.insert1(key)

make(key)

Populate the ArtifactDetection table.

If custom_artifact_detection is set in selection table, do nothing.

Fetches... - Parameters from ArtifactDetectionParameters - Recording from SpikeSortingRecording (loads with spikeinterface) Uses module-level function _get_artifact_times to detect artifacts.

Source code in src/spyglass/spikesorting/v0/spikesorting_artifact.py 
 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
157
def make(self, key):
    """Populate the ArtifactDetection table.

    If custom_artifact_detection is set in selection table, do nothing.

    Fetches...
        - Parameters from ArtifactDetectionParameters
        - Recording from SpikeSortingRecording (loads with spikeinterface)
    Uses module-level function _get_artifact_times to detect artifacts.
    """
    if not (ArtifactDetectionSelection & key).fetch1(
        "custom_artifact_detection"
    ):
        # get the dict of artifact params associated with this artifact_params_name
        artifact_params = (ArtifactDetectionParameters & key).fetch1(
            "artifact_params"
        )

        recording_path = (SpikeSortingRecording & key).fetch1(
            "recording_path"
        )
        recording_name = SpikeSortingRecording._get_recording_name(key)
        recording = si.load_extractor(recording_path)

        job_kwargs = {
            "chunk_duration": "10s",
            "n_jobs": 4,
            "progress_bar": "True",
        }

        artifact_removed_valid_times, artifact_times = _get_artifact_times(
            recording, **artifact_params, **job_kwargs
        )

        key["artifact_times"] = artifact_times
        key["artifact_removed_valid_times"] = artifact_removed_valid_times

        # set up a name for no-artifact times using recording id
        key["artifact_removed_interval_list_name"] = (
            recording_name
            + "_"
            + key["artifact_params_name"]
            + "_artifact_removed_valid_times"
        )

        ArtifactRemovedIntervalList.insert1(key, replace=True)

        # also insert into IntervalList
        tmp_key = {}
        tmp_key["nwb_file_name"] = key["nwb_file_name"]
        tmp_key["interval_list_name"] = key[
            "artifact_removed_interval_list_name"
        ]
        tmp_key["valid_times"] = key["artifact_removed_valid_times"]
        tmp_key["pipeline"] = "spikesorting_artifact_v0"
        IntervalList.insert1(tmp_key, replace=True)

        # insert into computed table
        self.insert1(key)