Skip to content

sortingview.py

SortingviewWorkspace

Bases: SpyglassMixin, Computed

Source code in src/spyglass/spikesorting/v0/sortingview.py 
 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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
@schema
class SortingviewWorkspace(SpyglassMixin, dj.Computed):
    definition = """
    -> SortingviewWorkspaceSelection
    ---
    workspace_uri: varchar(1000)
    sortingview_recording_id: varchar(30)
    sortingview_sorting_id: varchar(30)
    channel = NULL : varchar(80)        # the name of kachery channel for data sharing (for kachery daemon, deprecated)
    """

    # make class for parts table to hold URLs
    class URL(SpyglassMixin, dj.Part):
        # Table for holding URLs
        definition = """
        -> SortingviewWorkspace
        ---
        curation_url: varchar(1000)   # URL with sortingview data
        curation_jot: varchar(200)   # URI for saving manual curation tags
        """

    def make(self, key: dict):
        """Create a Sortingview workspace

        Parameters
        ----------
        key : dict
            primary key of an entry from SortingviewWorkspaceSelection table
        """

        # fetch
        recording_path = (SpikeSortingRecording & key).fetch1("recording_path")
        sorting_path = (SpikeSorting & key).fetch1("sorting_path")
        merge_groups = (Curation & key).fetch1("merge_groups")
        workspace_label = SpikeSortingRecording._get_recording_name(key)
        recording_label = SpikeSortingRecording._get_recording_name(key)
        sorting_label = SpikeSorting._get_sorting_name(key)
        metrics = (Curation & key).fetch1("quality_metrics")
        curation_labels = (Curation & key).fetch1("curation_labels")
        team_name = (SpikeSortingRecording & key).fetch1()["team_name"]
        team_members = (LabTeam.LabTeamMember & {"team_name": team_name}).fetch(
            "lab_member_name"
        )
        google_user_ids = []
        for team_member in team_members:
            google_user_id = (
                LabMember.LabMemberInfo & {"lab_member_name": team_member}
            ).fetch("google_user_name")
            if len(google_user_id) != 1:
                logger.warn(
                    f"Google user ID for {team_member} does not exist or more than one ID detected;\
                        permission not given to {team_member}, skipping..."
                )
                continue
            google_user_ids.append(google_user_id[0])

        # do
        (
            workspace_uri,
            recording_id,
            sorting_id,
        ) = _create_spikesortingview_workspace(
            recording_path=recording_path,
            sorting_path=sorting_path,
            merge_groups=merge_groups,
            workspace_label=workspace_label,
            recording_label=recording_label,
            sorting_label=sorting_label,
            metrics=metrics,
            curation_labels=curation_labels,
            google_user_ids=google_user_ids,
        )

        # insert
        key["workspace_uri"] = workspace_uri
        key["sortingview_recording_id"] = recording_id
        key["sortingview_sorting_id"] = sorting_id
        self.insert1(key)

        # insert URLs
        # remove non-primary keys
        del key["workspace_uri"]
        del key["sortingview_recording_id"]
        del key["sortingview_sorting_id"]

        # generate URLs and add to key
        url = self.url_trythis(key)
        key["curation_url"] = url
        key["curation_jot"] = "not ready yet"

        SortingviewWorkspace.URL.insert1(key)

    def remove_sorting_from_workspace(self, key):
        """Remove a sorting from the workspace. NOT IMPLEMENTED YET"""
        return NotImplementedError

    def url_trythis(self, key: dict, sortingview_sorting_id: str = None):
        """Generate a URL for visualizing and curating a sorting on the web.
        Will print instructions on how to do the curation.

        Parameters
        ----------
        key : dict
            An entry from SortingviewWorkspace table
        sortingview_sorting_id : str, optional
            sortingview sorting ID to visualize. If None then chooses the first one

        Returns
        -------
        url : str
        """
        workspace_uri = (self & key).fetch1("workspace_uri")
        workspace = sv.load_workspace(workspace_uri)
        recording_id = workspace.recording_ids[0]
        if sortingview_sorting_id is None:
            sortingview_sorting_id = workspace.sorting_ids[0]

        R = workspace.get_recording_extractor(recording_id)
        S = workspace.get_sorting_extractor(sortingview_sorting_id)

        initial_labels = (Curation & key).fetch("curation_labels")[0]
        for k, v in initial_labels.items():
            new_list = []
            for item in v:
                if item not in new_list:
                    new_list.append(item)
            initial_labels[k] = new_list
        initial_curation = {"labelsByUnit": initial_labels}

        # custom metrics
        unit_metrics = workspace.get_unit_metrics_for_sorting(
            sortingview_sorting_id
        )

        # This will print some instructions on how to do the curation
        # old: sv.trythis_start_sorting_curation
        url = _generate_url(
            recording=R,
            sorting=S,
            label=workspace.label,
            initial_curation=initial_curation,
            raster_plot_subsample_max_firing_rate=50,
            spike_amplitudes_subsample_max_firing_rate=50,
            unit_metrics=unit_metrics,
        )
        return url

    def insert_manual_curation(
        self, key: dict, url: str, description="manually curated"
    ):
        """Based on information in key for an SortingviewWorkspace, loads the
        curated sorting from sortingview, saves it (with labels and the
        optional description) and inserts it to CuratedSorting

        Assumes that the workspace corresponding to the recording and (original) sorting exists

        Parameters
        ----------
        key : dict
            primary key of AutomaticCuration
        description: str, optional
            description of curated sorting
        """

        # get the labels and remove the non-primary merged units
        # labels = workspace.get_sorting_curation(sorting_id=sortingview_sorting_id)
        # labels = sv.trythis_load_sorting_curation('jot://xTzzyDieQPkW')
        labels = sv.trythis_load_sorting_curation(url)

        # turn labels to list of str, only including accepted units.
        # if bool(labels["mergeGroups"]):
        if bool(labels.get("mergeGroups", [])):
            # clusters were merged, so we empty out metrics
            metrics = {}
        else:
            # get the metrics from the parent curation
            metrics = (Curation & key).fetch1("quality_metrics")

        # insert this curation into the  Table
        return Curation.insert_curation(
            key,
            parent_curation_id=key["curation_id"],
            labels=labels["labelsByUnit"],
            merge_groups=labels.get("mergeGroups", []),
            metrics=metrics,
            description=description,
        )

make(key)

Create a Sortingview workspace

Parameters:

Name Type Description Default
key dict

primary key of an entry from SortingviewWorkspaceSelection table

 required
Source code in src/spyglass/spikesorting/v0/sortingview.py 
 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
def make(self, key: dict):
    """Create a Sortingview workspace

    Parameters
    ----------
    key : dict
        primary key of an entry from SortingviewWorkspaceSelection table
    """

    # fetch
    recording_path = (SpikeSortingRecording & key).fetch1("recording_path")
    sorting_path = (SpikeSorting & key).fetch1("sorting_path")
    merge_groups = (Curation & key).fetch1("merge_groups")
    workspace_label = SpikeSortingRecording._get_recording_name(key)
    recording_label = SpikeSortingRecording._get_recording_name(key)
    sorting_label = SpikeSorting._get_sorting_name(key)
    metrics = (Curation & key).fetch1("quality_metrics")
    curation_labels = (Curation & key).fetch1("curation_labels")
    team_name = (SpikeSortingRecording & key).fetch1()["team_name"]
    team_members = (LabTeam.LabTeamMember & {"team_name": team_name}).fetch(
        "lab_member_name"
    )
    google_user_ids = []
    for team_member in team_members:
        google_user_id = (
            LabMember.LabMemberInfo & {"lab_member_name": team_member}
        ).fetch("google_user_name")
        if len(google_user_id) != 1:
            logger.warn(
                f"Google user ID for {team_member} does not exist or more than one ID detected;\
                    permission not given to {team_member}, skipping..."
            )
            continue
        google_user_ids.append(google_user_id[0])

    # do
    (
        workspace_uri,
        recording_id,
        sorting_id,
    ) = _create_spikesortingview_workspace(
        recording_path=recording_path,
        sorting_path=sorting_path,
        merge_groups=merge_groups,
        workspace_label=workspace_label,
        recording_label=recording_label,
        sorting_label=sorting_label,
        metrics=metrics,
        curation_labels=curation_labels,
        google_user_ids=google_user_ids,
    )

    # insert
    key["workspace_uri"] = workspace_uri
    key["sortingview_recording_id"] = recording_id
    key["sortingview_sorting_id"] = sorting_id
    self.insert1(key)

    # insert URLs
    # remove non-primary keys
    del key["workspace_uri"]
    del key["sortingview_recording_id"]
    del key["sortingview_sorting_id"]

    # generate URLs and add to key
    url = self.url_trythis(key)
    key["curation_url"] = url
    key["curation_jot"] = "not ready yet"

    SortingviewWorkspace.URL.insert1(key)

remove_sorting_from_workspace(key)

Remove a sorting from the workspace. NOT IMPLEMENTED YET

Source code in src/spyglass/spikesorting/v0/sortingview.py 
118
119
120
def remove_sorting_from_workspace(self, key):
    """Remove a sorting from the workspace. NOT IMPLEMENTED YET"""
    return NotImplementedError

url_trythis(key, sortingview_sorting_id=None)

Generate a URL for visualizing and curating a sorting on the web. Will print instructions on how to do the curation.

Parameters:

Name Type Description Default
key dict

An entry from SortingviewWorkspace table

 required
sortingview_sorting_id str

sortingview sorting ID to visualize. If None then chooses the first one

 None

Returns:

Name Type Description
url str
Source code in src/spyglass/spikesorting/v0/sortingview.py 
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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def url_trythis(self, key: dict, sortingview_sorting_id: str = None):
    """Generate a URL for visualizing and curating a sorting on the web.
    Will print instructions on how to do the curation.

    Parameters
    ----------
    key : dict
        An entry from SortingviewWorkspace table
    sortingview_sorting_id : str, optional
        sortingview sorting ID to visualize. If None then chooses the first one

    Returns
    -------
    url : str
    """
    workspace_uri = (self & key).fetch1("workspace_uri")
    workspace = sv.load_workspace(workspace_uri)
    recording_id = workspace.recording_ids[0]
    if sortingview_sorting_id is None:
        sortingview_sorting_id = workspace.sorting_ids[0]

    R = workspace.get_recording_extractor(recording_id)
    S = workspace.get_sorting_extractor(sortingview_sorting_id)

    initial_labels = (Curation & key).fetch("curation_labels")[0]
    for k, v in initial_labels.items():
        new_list = []
        for item in v:
            if item not in new_list:
                new_list.append(item)
        initial_labels[k] = new_list
    initial_curation = {"labelsByUnit": initial_labels}

    # custom metrics
    unit_metrics = workspace.get_unit_metrics_for_sorting(
        sortingview_sorting_id
    )

    # This will print some instructions on how to do the curation
    # old: sv.trythis_start_sorting_curation
    url = _generate_url(
        recording=R,
        sorting=S,
        label=workspace.label,
        initial_curation=initial_curation,
        raster_plot_subsample_max_firing_rate=50,
        spike_amplitudes_subsample_max_firing_rate=50,
        unit_metrics=unit_metrics,
    )
    return url

insert_manual_curation(key, url, description='manually curated')

Based on information in key for an SortingviewWorkspace, loads the curated sorting from sortingview, saves it (with labels and the optional description) and inserts it to CuratedSorting

Assumes that the workspace corresponding to the recording and (original) sorting exists

Parameters:

Name Type Description Default
key dict

primary key of AutomaticCuration

 required
description

description of curated sorting

 'manually curated'
Source code in src/spyglass/spikesorting/v0/sortingview.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def insert_manual_curation(
    self, key: dict, url: str, description="manually curated"
):
    """Based on information in key for an SortingviewWorkspace, loads the
    curated sorting from sortingview, saves it (with labels and the
    optional description) and inserts it to CuratedSorting

    Assumes that the workspace corresponding to the recording and (original) sorting exists

    Parameters
    ----------
    key : dict
        primary key of AutomaticCuration
    description: str, optional
        description of curated sorting
    """

    # get the labels and remove the non-primary merged units
    # labels = workspace.get_sorting_curation(sorting_id=sortingview_sorting_id)
    # labels = sv.trythis_load_sorting_curation('jot://xTzzyDieQPkW')
    labels = sv.trythis_load_sorting_curation(url)

    # turn labels to list of str, only including accepted units.
    # if bool(labels["mergeGroups"]):
    if bool(labels.get("mergeGroups", [])):
        # clusters were merged, so we empty out metrics
        metrics = {}
    else:
        # get the metrics from the parent curation
        metrics = (Curation & key).fetch1("quality_metrics")

    # insert this curation into the  Table
    return Curation.insert_curation(
        key,
        parent_curation_id=key["curation_id"],
        labels=labels["labelsByUnit"],
        merge_groups=labels.get("mergeGroups", []),
        metrics=metrics,
        description=description,
    )