Skip to content

utils.py

generate_nwb_uuid(nwb_file_name, initial, len_uuid=6)

Generates a unique identifier related to an NWB file.

Parameters:

Name Type Description Default
nwb_file_name str

Nwb file name, first part of resulting string.

 required
initial str

R if recording; A if artifact; S if sorting etc

 required
len_uuid int

how many digits of uuid4 to keep

 6

Returns:

Type Description
str

A unique identifier for the NWB file. "{nwbf}{initial}"
Source code in src/spyglass/spikesorting/v1/utils.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def generate_nwb_uuid(
    nwb_file_name: str, initial: str, len_uuid: int = 6
) -> str:
    """Generates a unique identifier related to an NWB file.

    Parameters
    ----------
    nwb_file_name : str
        Nwb file name, first part of resulting string.
    initial : str
        R if recording; A if artifact; S if sorting etc
    len_uuid : int
        how many digits of uuid4 to keep

    Returns
    -------
    str
        A unique identifier for the NWB file.
        "{nwbf}_{initial}_{uuid4[:len_uuid]}"
    """
    uuid4 = str(uuid.uuid4())
    nwb_uuid = nwb_file_name + "_" + initial + "_" + uuid4[:len_uuid]
    return nwb_uuid

get_spiking_sorting_v1_merge_ids(restriction)

Parses the SpikingSorting V1 pipeline to get a list of merge ids for a given restriction.

Parameters:

Name Type Description Default
restriction dict

A dictionary containing some or all of the following key-value pairs: nwb_file_name : str name of the nwb file interval_list_name : str name of the interval list sort_group_name : str name of the sort group artifact_param_name : str name of the artifact parameter curation_id : int, optional id of the curation (if not specified, uses the latest curation)

 required

Returns:

Name Type Description
merge_id_list list

list of merge ids for the given restriction
Source code in src/spyglass/spikesorting/v1/utils.py 
 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
def get_spiking_sorting_v1_merge_ids(restriction: dict):
    """
    Parses the SpikingSorting V1 pipeline to get a list of merge ids for a given restriction.

    Parameters
    ----------
    restriction : dict
        A dictionary containing some or all of the following key-value pairs:
        nwb_file_name : str
            name of the nwb file
        interval_list_name : str
            name of the interval list
        sort_group_name : str
            name of the sort group
        artifact_param_name : str
            name of the artifact parameter
        curation_id : int, optional
            id of the curation (if not specified, uses the latest curation)

    Returns
    -------
    merge_id_list : list
        list of merge ids for the given restriction
    """
    # list of recording ids
    recording_id_list = (SpikeSortingRecordingSelection() & restriction).fetch(
        "recording_id"
    )
    # list of artifact ids for each recording
    artifact_id_list = [
        (
            ArtifactDetectionSelection() & restriction & {"recording_id": id}
        ).fetch1("artifact_id")
        for id in recording_id_list
    ]
    # list of sorting ids for each recording
    sorting_restriction = restriction.copy()
    _ = sorting_restriction.pop("interval_list_name", None)
    sorting_id_list = []
    for r_id, a_id in zip(recording_id_list, artifact_id_list):
        rec_dict = {"recording_id": str(r_id), "interval_list_name": str(a_id)}
        # if sorted with artifact detection
        if SpikeSortingSelection() & sorting_restriction & rec_dict:
            sorting_id_list.append(
                (
                    SpikeSortingSelection() & sorting_restriction & rec_dict
                ).fetch1("sorting_id")
            )
        # if sorted without artifact detection
        else:
            sorting_id_list.append(
                (
                    SpikeSortingSelection() & sorting_restriction & rec_dict
                ).fetch1("sorting_id")
            )
    # if curation_id is specified, use that id for each sorting_id
    if "curation_id" in restriction:
        curation_id = [restriction["curation_id"] for _ in sorting_id_list]
    # if curation_id is not specified, use the latest curation_id for each sorting_id
    else:
        curation_id = [
            np.max((CurationV1 & {"sorting_id": id}).fetch("curation_id"))
            for id in sorting_id_list
        ]
    # list of merge ids for the desired curation(s)
    merge_id_list = [
        (
            SpikeSortingOutput.CurationV1()
            & {"sorting_id": s_id, "curation_id": c_id}
        ).fetch1("merge_id")
        for s_id, c_id in zip(sorting_id_list, curation_id)
    ]
    return merge_id_list