Skip to content

common_lab.py

Schema for institution, lab team/name/members. Session-independent.

LabMember

Bases: SpyglassMixin, Manual

Source code in src/spyglass/common/common_lab.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 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
@schema
class LabMember(SpyglassMixin, dj.Manual):
    definition = """
    lab_member_name: varchar(80)
    ---
    first_name: varchar(200)
    last_name: varchar(200)
    """

    # NOTE that names must be unique here. If there are two neuroscientists
    # named Jack Black that have data in this database, this will create an
    # incorrect linkage. NWB does not yet provide unique IDs for names.

    # NOTE: PR requires table alter.
    class LabMemberInfo(SpyglassMixin, dj.Part):
        definition = """
        # Information about lab member in the context of Frank lab network
        -> LabMember
        ---
        google_user_name         : varchar(200) # For permission to curate
        datajoint_user_name = "" : varchar(200) # For permission to delete
        admin = 0                : bool         # Ignore permission checks
        unique index (datajoint_user_name)
        unique index (google_user_name)
        """

        # NOTE: index present for new instances. pending surgery, not enforced
        # for existing instances.

    _admin = []

    @classmethod
    def insert_from_nwbfile(cls, nwbf, config=None):
        """Insert lab member information from an NWB file.

        Parameters
        ----------
        nwbf: pynwb.NWBFile
            The NWB file with experimenter information.
        config : dict
            Dictionary read from a user-defined YAML file containing values to
            replace in the NWB file.
        """
        config = config or dict()
        if isinstance(nwbf, str):
            nwb_file_abspath = Nwbfile.get_abs_path(nwbf, new_file=True)
            nwbf = get_nwb_file(nwb_file_abspath)

        if "LabMember" in config:
            experimenter_list = [
                member_dict["lab_member_name"]
                for member_dict in config["LabMember"]
            ]
        elif nwbf.experimenter is not None:
            experimenter_list = nwbf.experimenter
        else:
            logger.info("No experimenter metadata found.\n")
            return

        for experimenter in experimenter_list:
            cls.insert_from_name(experimenter)

            # each person is by default the member of their own LabTeam
            # (same as their name)

            full_name, first, last = decompose_name(experimenter)
            LabTeam.create_new_team(
                team_name=full_name, team_members=[f"{last}, {first}"]
            )

    @classmethod
    def insert_from_name(cls, full_name):
        """Insert a lab member by name.

        Parameters
        ----------
        full_name : str
            The name to be added.
        """
        _, first, last = decompose_name(full_name)
        cls.insert1(
            dict(
                lab_member_name=f"{first} {last}",
                first_name=first,
                last_name=last,
            ),
            skip_duplicates=True,
        )

    def _load_admin(cls):
        """Load admin list."""
        cls._admin = list(
            (cls.LabMemberInfo & {"admin": True}).fetch("datajoint_user_name")
        ) + ["root"]

    @property
    def admin(cls) -> list:
        """Return the list of admin users.

        Note: This is cached. If adding a new admin, run _load_admin
        """
        if not cls._admin:
            cls._load_admin()
        return cls._admin

    def get_djuser_name(cls, dj_user) -> str:
        """Return the lab member name associated with a datajoint user name.

        Parameters
        ----------
        dj_user: str
            The datajoint user name.

        Returns
        -------
        str
            The lab member name.
        """
        query = (cls.LabMemberInfo & {"datajoint_user_name": dj_user}).fetch(
            "lab_member_name"
        )

        if len(query) != 1:
            raise ValueError(
                f"Could not find name for datajoint user {dj_user}"
                + f" in common.LabMember.LabMemberInfo: {query}"
            )

        return query[0]

    def check_admin_privilege(
        cls,
        error_message: str = "User does not have database admin privileges",
    ):
        """Check if a user has admin privilege.

        Parameters
        ----------
        error_message: str
            The error message to display if the user is not an admin.
        """
        if dj.config["database.user"] not in cls.admin:
            raise PermissionError(error_message)

insert_from_nwbfile(nwbf, config=None) classmethod

Insert lab member information from an NWB file.

Parameters:

Name Type Description Default
nwbf

The NWB file with experimenter information.

 required
config dict

Dictionary read from a user-defined YAML file containing values to replace in the NWB file.

 None
Source code in src/spyglass/common/common_lab.py 
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
@classmethod
def insert_from_nwbfile(cls, nwbf, config=None):
    """Insert lab member information from an NWB file.

    Parameters
    ----------
    nwbf: pynwb.NWBFile
        The NWB file with experimenter information.
    config : dict
        Dictionary read from a user-defined YAML file containing values to
        replace in the NWB file.
    """
    config = config or dict()
    if isinstance(nwbf, str):
        nwb_file_abspath = Nwbfile.get_abs_path(nwbf, new_file=True)
        nwbf = get_nwb_file(nwb_file_abspath)

    if "LabMember" in config:
        experimenter_list = [
            member_dict["lab_member_name"]
            for member_dict in config["LabMember"]
        ]
    elif nwbf.experimenter is not None:
        experimenter_list = nwbf.experimenter
    else:
        logger.info("No experimenter metadata found.\n")
        return

    for experimenter in experimenter_list:
        cls.insert_from_name(experimenter)

        # each person is by default the member of their own LabTeam
        # (same as their name)

        full_name, first, last = decompose_name(experimenter)
        LabTeam.create_new_team(
            team_name=full_name, team_members=[f"{last}, {first}"]
        )

insert_from_name(full_name) classmethod

Insert a lab member by name.

Parameters:

Name Type Description Default
full_name str

The name to be added.

 required
Source code in src/spyglass/common/common_lab.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@classmethod
def insert_from_name(cls, full_name):
    """Insert a lab member by name.

    Parameters
    ----------
    full_name : str
        The name to be added.
    """
    _, first, last = decompose_name(full_name)
    cls.insert1(
        dict(
            lab_member_name=f"{first} {last}",
            first_name=first,
            last_name=last,
        ),
        skip_duplicates=True,
    )

admin: list property

Return the list of admin users.

Note: This is cached. If adding a new admin, run _load_admin

get_djuser_name(dj_user)

Return the lab member name associated with a datajoint user name.

Parameters:

Name Type Description Default
dj_user

The datajoint user name.

 required

Returns:

Type Description
str

The lab member name.
Source code in src/spyglass/common/common_lab.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def get_djuser_name(cls, dj_user) -> str:
    """Return the lab member name associated with a datajoint user name.

    Parameters
    ----------
    dj_user: str
        The datajoint user name.

    Returns
    -------
    str
        The lab member name.
    """
    query = (cls.LabMemberInfo & {"datajoint_user_name": dj_user}).fetch(
        "lab_member_name"
    )

    if len(query) != 1:
        raise ValueError(
            f"Could not find name for datajoint user {dj_user}"
            + f" in common.LabMember.LabMemberInfo: {query}"
        )

    return query[0]

check_admin_privilege(error_message='User does not have database admin privileges')

Check if a user has admin privilege.

Parameters:

Name Type Description Default
error_message str

The error message to display if the user is not an admin.

 'User does not have database admin privileges'
Source code in src/spyglass/common/common_lab.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
def check_admin_privilege(
    cls,
    error_message: str = "User does not have database admin privileges",
):
    """Check if a user has admin privilege.

    Parameters
    ----------
    error_message: str
        The error message to display if the user is not an admin.
    """
    if dj.config["database.user"] not in cls.admin:
        raise PermissionError(error_message)

LabTeam

Bases: SpyglassMixin, Manual

Source code in src/spyglass/common/common_lab.py 
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
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@schema
class LabTeam(SpyglassMixin, dj.Manual):
    definition = """
    team_name: varchar(80)
    ---
    team_description = "": varchar(2000)
    """

    class LabTeamMember(SpyglassMixin, dj.Part):
        definition = """
        -> LabTeam
        -> LabMember
        """

    _shared_teams = {}

    def get_team_members(cls, member, reload=False):
        """Return the set of lab members that share a team with member.

        Parameters
        ----------
        member : str
            The lab member to check.
        reload : bool
            If True, reload the shared teams cache.
        """
        if reload or member not in cls._shared_teams:
            teams = (cls.LabTeamMember & {"lab_member_name": member}).fetch(
                "team_name"
            )
            cls._shared_teams[member] = set(
                (
                    LabTeam.LabTeamMember
                    & [{"team_name": team} for team in teams]
                ).fetch("lab_member_name")
            )
        return cls._shared_teams[member]

    @classmethod
    def create_new_team(
        cls, team_name: str, team_members: list, team_description: str = ""
    ):
        """Create a new team with a list of team members.

        If the lab member does not exist in the database, they will be added.

        Parameters
        ----------
        team_name : str
            The name of the team.
        team_members : str
            The full names of the lab members that are part of the team.
        team_description: str
            The description of the team.
        """
        labteam_dict = {
            "team_name": team_name,
            "team_description": team_description,
        }

        member_list = []
        for team_member in team_members:
            LabMember.insert_from_name(team_member)
            member_dict = {"lab_member_name": decompose_name(team_member)[0]}
            query = (LabMember.LabMemberInfo() & member_dict).fetch(
                "google_user_name"
            )
            if not query:
                logger.info(
                    "To help manage permissions in LabMemberInfo, please add Google "
                    + f"user ID for {team_member}"
                )
            labteammember_dict = {
                "team_name": team_name,
                **member_dict,
            }
            member_list.append(labteammember_dict)
            # clear cache for this member
            _ = cls._shared_teams.pop(team_member, None)

        cls.insert1(labteam_dict, skip_duplicates=True)
        cls.LabTeamMember.insert(member_list, skip_duplicates=True)

get_team_members(member, reload=False)

Return the set of lab members that share a team with member.

Parameters:

Name Type Description Default
member str

The lab member to check.

 required
reload bool

If True, reload the shared teams cache.

 False
Source code in src/spyglass/common/common_lab.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def get_team_members(cls, member, reload=False):
    """Return the set of lab members that share a team with member.

    Parameters
    ----------
    member : str
        The lab member to check.
    reload : bool
        If True, reload the shared teams cache.
    """
    if reload or member not in cls._shared_teams:
        teams = (cls.LabTeamMember & {"lab_member_name": member}).fetch(
            "team_name"
        )
        cls._shared_teams[member] = set(
            (
                LabTeam.LabTeamMember
                & [{"team_name": team} for team in teams]
            ).fetch("lab_member_name")
        )
    return cls._shared_teams[member]

create_new_team(team_name, team_members, team_description='') classmethod

Create a new team with a list of team members.

If the lab member does not exist in the database, they will be added.

Parameters:

Name Type Description Default
team_name str

The name of the team.

 required
team_members str

The full names of the lab members that are part of the team.

 required
team_description str

The description of the team.

 ''
Source code in src/spyglass/common/common_lab.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@classmethod
def create_new_team(
    cls, team_name: str, team_members: list, team_description: str = ""
):
    """Create a new team with a list of team members.

    If the lab member does not exist in the database, they will be added.

    Parameters
    ----------
    team_name : str
        The name of the team.
    team_members : str
        The full names of the lab members that are part of the team.
    team_description: str
        The description of the team.
    """
    labteam_dict = {
        "team_name": team_name,
        "team_description": team_description,
    }

    member_list = []
    for team_member in team_members:
        LabMember.insert_from_name(team_member)
        member_dict = {"lab_member_name": decompose_name(team_member)[0]}
        query = (LabMember.LabMemberInfo() & member_dict).fetch(
            "google_user_name"
        )
        if not query:
            logger.info(
                "To help manage permissions in LabMemberInfo, please add Google "
                + f"user ID for {team_member}"
            )
        labteammember_dict = {
            "team_name": team_name,
            **member_dict,
        }
        member_list.append(labteammember_dict)
        # clear cache for this member
        _ = cls._shared_teams.pop(team_member, None)

    cls.insert1(labteam_dict, skip_duplicates=True)
    cls.LabTeamMember.insert(member_list, skip_duplicates=True)

Institution

Bases: SpyglassMixin, Manual

Source code in src/spyglass/common/common_lab.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
@schema
class Institution(SpyglassMixin, dj.Manual):
    definition = """
    institution_name: varchar(80)
    """

    @classmethod
    def insert_from_nwbfile(cls, nwbf, config=None):
        """Insert institution information from an NWB file.

        Parameters
        ----------
        nwbf : pynwb.NWBFile
            The NWB file with institution information.
        config : dict
            Dictionary read from a user-defined YAML file containing values to
            replace in the NWB file.

        Returns
        -------
        institution_name : string
            The name of the institution found in the NWB or config file, or None.
        """
        config = config or dict()
        inst_list = config.get("Institution", [{}])
        if len(inst_list) > 1:
            logger.info(
                "Multiple institution entries not allowed. Using the first entry only.\n"
            )
        inst_name = inst_list[0].get("institution_name") or getattr(
            nwbf, "institution", None
        )
        if not inst_name:
            logger.info("No institution metadata found.\n")
            return None

        cls.insert1(dict(institution_name=inst_name), skip_duplicates=True)
        return inst_name

insert_from_nwbfile(nwbf, config=None) classmethod

Insert institution information from an NWB file.

Parameters:

Name Type Description Default
nwbf NWBFile

The NWB file with institution information.

 required
config dict

Dictionary read from a user-defined YAML file containing values to replace in the NWB file.

 None

Returns:

Name Type Description
institution_name string

The name of the institution found in the NWB or config file, or None.
Source code in src/spyglass/common/common_lab.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
@classmethod
def insert_from_nwbfile(cls, nwbf, config=None):
    """Insert institution information from an NWB file.

    Parameters
    ----------
    nwbf : pynwb.NWBFile
        The NWB file with institution information.
    config : dict
        Dictionary read from a user-defined YAML file containing values to
        replace in the NWB file.

    Returns
    -------
    institution_name : string
        The name of the institution found in the NWB or config file, or None.
    """
    config = config or dict()
    inst_list = config.get("Institution", [{}])
    if len(inst_list) > 1:
        logger.info(
            "Multiple institution entries not allowed. Using the first entry only.\n"
        )
    inst_name = inst_list[0].get("institution_name") or getattr(
        nwbf, "institution", None
    )
    if not inst_name:
        logger.info("No institution metadata found.\n")
        return None

    cls.insert1(dict(institution_name=inst_name), skip_duplicates=True)
    return inst_name

Lab

Bases: SpyglassMixin, Manual

Source code in src/spyglass/common/common_lab.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
@schema
class Lab(SpyglassMixin, dj.Manual):
    definition = """
    lab_name: varchar(80)
    """

    @classmethod
    def insert_from_nwbfile(cls, nwbf, config=None):
        """Insert lab name information from an NWB file.

        Parameters
        ----------
        nwbf : pynwb.NWBFile
            The NWB file with lab name information.
        config : dict
            Dictionary read from a user-defined YAML file containing values to
            replace in the NWB file.

        Returns
        -------
        lab_name : string
            The name of the lab found in the NWB or config file, or None.
        """
        config = config or dict()
        lab_list = config.get("Lab", [{}])
        if len(lab_list) > 1:
            logger.info(
                "Multiple lab entries not allowed. Using the first entry only.\n"
            )
        lab_name = lab_list[0].get("lab_name") or getattr(nwbf, "lab", None)
        if not lab_name:
            logger.info("No lab metadata found.\n")
            return None

        cls.insert1(dict(lab_name=lab_name), skip_duplicates=True)
        return lab_name

insert_from_nwbfile(nwbf, config=None) classmethod

Insert lab name information from an NWB file.

Parameters:

Name Type Description Default
nwbf NWBFile

The NWB file with lab name information.

 required
config dict

Dictionary read from a user-defined YAML file containing values to replace in the NWB file.

 None

Returns:

Name Type Description
lab_name string

The name of the lab found in the NWB or config file, or None.
Source code in src/spyglass/common/common_lab.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
@classmethod
def insert_from_nwbfile(cls, nwbf, config=None):
    """Insert lab name information from an NWB file.

    Parameters
    ----------
    nwbf : pynwb.NWBFile
        The NWB file with lab name information.
    config : dict
        Dictionary read from a user-defined YAML file containing values to
        replace in the NWB file.

    Returns
    -------
    lab_name : string
        The name of the lab found in the NWB or config file, or None.
    """
    config = config or dict()
    lab_list = config.get("Lab", [{}])
    if len(lab_list) > 1:
        logger.info(
            "Multiple lab entries not allowed. Using the first entry only.\n"
        )
    lab_name = lab_list[0].get("lab_name") or getattr(nwbf, "lab", None)
    if not lab_name:
        logger.info("No lab metadata found.\n")
        return None

    cls.insert1(dict(lab_name=lab_name), skip_duplicates=True)
    return lab_name

decompose_name(full_name)

Return the full, first and last name parts of a full name.

Names capitalized. Decomposes either comma- or space-separated. If name includes spaces, must use exactly one comma.

Parameters:

Name Type Description Default
full_name str

The full name to decompose.

 required

Returns:

Type Description
tuple

A tuple of the full, first, last name parts.

Raises:

Type Description
ValueError

When full_name is in an unsupported format
Source code in src/spyglass/common/common_lab.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def decompose_name(full_name: str) -> tuple:
    """Return the full, first and last name parts of a full name.

    Names capitalized. Decomposes either comma- or space-separated.
    If name includes spaces, must use exactly one comma.

    Parameters
    ----------
    full_name: str
        The full name to decompose.

    Returns
    -------
    tuple
        A tuple of the full, first, last name parts.

    Raises
    ------
    ValueError
        When full_name is in an unsupported format
    """

    full_trimmed = "" if not full_name else full_name.strip()

    delim = (
        ", "
        if full_trimmed.count(", ") == 1
        else " " if full_trimmed.count(" ") == 1 else None
    )

    parts = full_trimmed.split(delim)

    if delim is None or len(parts) != 2:  # catch unsupported format
        raise ValueError(
            f"Name has unsupported format for {full_name}. \n\t"
            + "Must use exactly one comma+space (i.e., ', ') or space.\n\t"
            + "And provide exactly two name parts.\n\t"
            + "For example, 'First Last' or 'Last1 Last2, First1 First2' "
            + "if name(s) includes spaces."
        )

    first, last = parts if delim == " " else parts[::-1]
    full = f"{first} {last}"

    return full, first, last