Skip to content

common_task.py

Task

Bases: SpyglassMixin, Manual

Source code in src/spyglass/common/common_task.py 
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
@schema
class Task(SpyglassMixin, dj.Manual):
    definition = """
     task_name: varchar(80)
     ---
     task_description = NULL: varchar(2000)    # description of this task
     task_type = NULL: varchar(2000)           # type of task
     task_subtype = NULL: varchar(2000)        # subtype of task
     """

    @classmethod
    def insert_from_nwbfile(cls, nwbf):
        """Insert tasks from an NWB file.

        Parameters
        ----------
        nwbf : pynwb.NWBFile
            The source NWB file object.
        """
        tasks_mod = nwbf.processing.get("tasks")
        if tasks_mod is None:
            logger.warn(f"No tasks processing module found in {nwbf}\n")
            return
        for task in tasks_mod.data_interfaces.values():
            if cls.check_task_table(task):
                cls.insert_from_task_table(task)

    @classmethod
    def insert_from_task_table(cls, task_table):
        """Insert tasks from a pynwb DynamicTable containing task metadata.

        Duplicate tasks will not be added.

        Parameters
        ----------
        task_table : pynwb.core.DynamicTable
            The table representing task metadata.
        """
        taskdf = task_table.to_dataframe()
        for task_entry in taskdf.iterrows():
            task_dict = dict()
            task_dict["task_name"] = task_entry[1].task_name
            task_dict["task_description"] = task_entry[1].task_description
            cls.insert1(task_dict, skip_duplicates=True)

    @classmethod
    def check_task_table(cls, task_table):
        """Check whether the pynwb DynamicTable containing task metadata conforms to the expected format.


        The table should be an instance of pynwb.core.DynamicTable and contain the columns 'task_name' and
        'task_description'.

        Parameters
        ----------
        task_table : pynwb.core.DynamicTable
            The table representing task metadata.

        Returns
        -------
        bool
            Whether the DynamicTable conforms to the expected format for loading data into the Task table.
        """
        return (
            isinstance(task_table, pynwb.core.DynamicTable)
            and hasattr(task_table, "task_name")
            and hasattr(task_table, "task_description")
        )

insert_from_nwbfile(nwbf) classmethod

Insert tasks from an NWB file.

Parameters:

Name Type Description Default
nwbf NWBFile

The source NWB file object.

 required
Source code in src/spyglass/common/common_task.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@classmethod
def insert_from_nwbfile(cls, nwbf):
    """Insert tasks from an NWB file.

    Parameters
    ----------
    nwbf : pynwb.NWBFile
        The source NWB file object.
    """
    tasks_mod = nwbf.processing.get("tasks")
    if tasks_mod is None:
        logger.warn(f"No tasks processing module found in {nwbf}\n")
        return
    for task in tasks_mod.data_interfaces.values():
        if cls.check_task_table(task):
            cls.insert_from_task_table(task)

insert_from_task_table(task_table) classmethod

Insert tasks from a pynwb DynamicTable containing task metadata.

Duplicate tasks will not be added.

Parameters:

Name Type Description Default
task_table DynamicTable

The table representing task metadata.

 required
Source code in src/spyglass/common/common_task.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@classmethod
def insert_from_task_table(cls, task_table):
    """Insert tasks from a pynwb DynamicTable containing task metadata.

    Duplicate tasks will not be added.

    Parameters
    ----------
    task_table : pynwb.core.DynamicTable
        The table representing task metadata.
    """
    taskdf = task_table.to_dataframe()
    for task_entry in taskdf.iterrows():
        task_dict = dict()
        task_dict["task_name"] = task_entry[1].task_name
        task_dict["task_description"] = task_entry[1].task_description
        cls.insert1(task_dict, skip_duplicates=True)

check_task_table(task_table) classmethod

Check whether the pynwb DynamicTable containing task metadata conforms to the expected format.

The table should be an instance of pynwb.core.DynamicTable and contain the columns 'task_name' and 'task_description'.

Parameters:

Name Type Description Default
task_table DynamicTable

The table representing task metadata.

 required

Returns:

Type Description
bool

Whether the DynamicTable conforms to the expected format for loading data into the Task table.
Source code in src/spyglass/common/common_task.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@classmethod
def check_task_table(cls, task_table):
    """Check whether the pynwb DynamicTable containing task metadata conforms to the expected format.


    The table should be an instance of pynwb.core.DynamicTable and contain the columns 'task_name' and
    'task_description'.

    Parameters
    ----------
    task_table : pynwb.core.DynamicTable
        The table representing task metadata.

    Returns
    -------
    bool
        Whether the DynamicTable conforms to the expected format for loading data into the Task table.
    """
    return (
        isinstance(task_table, pynwb.core.DynamicTable)
        and hasattr(task_table, "task_name")
        and hasattr(task_table, "task_description")
    )

TaskEpoch

Bases: SpyglassMixin, Imported

Source code in src/spyglass/common/common_task.py 
 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
213
214
@schema
class TaskEpoch(SpyglassMixin, dj.Imported):
    # Tasks, session and time intervals
    definition = """
     -> Session
     epoch: int  # the session epoch for this task and apparatus(1 based)
     ---
     -> Task
     -> [nullable] CameraDevice
     -> IntervalList
     task_environment = NULL: varchar(200)  # the environment the animal was in
     camera_names : blob # list of keys corresponding to entry in CameraDevice
     """

    def make(self, key):
        nwb_file_name = key["nwb_file_name"]
        nwb_file_abspath = Nwbfile().get_abs_path(nwb_file_name)
        nwbf = get_nwb_file(nwb_file_abspath)
        camera_names = dict()

        # the tasks refer to the camera_id which is unique for the NWB file but
        # not for CameraDevice schema, so we need to look up the right camera
        # map camera ID (in camera name) to camera_name

        for device in nwbf.devices.values():
            if isinstance(device, ndx_franklab_novela.CameraDevice):
                # get the camera ID
                camera_id = int(str.split(device.name)[1])
                camera_names[camera_id] = device.camera_name

        # find the task modules and for each one, add the task to the Task
        # schema if it isn't there and then add an entry for each epoch

        tasks_mod = nwbf.processing.get("tasks")
        if tasks_mod is None:
            logger.warn(f"No tasks processing module found in {nwbf}\n")
            return

        task_inserts = []
        for task in tasks_mod.data_interfaces.values():
            if self.check_task_table(task):
                # check if the task is in the Task table and if not, add it
                Task.insert_from_task_table(task)
                key["task_name"] = task.task_name[0]

                # get the CameraDevice used for this task (primary key is
                # camera name so we need to map from ID to name)

                camera_ids = task.camera_id[0]
                valid_camera_ids = [
                    camera_id
                    for camera_id in camera_ids
                    if camera_id in camera_names.keys()
                ]
                if valid_camera_ids:
                    key["camera_names"] = [
                        {"camera_name": camera_names[camera_id]}
                        for camera_id in valid_camera_ids
                    ]
                else:
                    logger.warn(
                        f"No camera device found with ID {camera_ids} in NWB "
                        + f"file {nwbf}\n"
                    )
                # Add task environment
                if hasattr(task, "task_environment"):
                    key["task_environment"] = task.task_environment[0]

                # get the interval list for this task, which corresponds to the
                # matching epoch for the raw data. Users should define more
                # restrictive intervals as required for analyses

                session_intervals = (
                    IntervalList() & {"nwb_file_name": nwb_file_name}
                ).fetch("interval_list_name")
                for epoch in task.task_epochs[0]:
                    # TODO in beans file, task_epochs[0] is 1x2 dset of ints,
                    # so epoch would be an int

                    key["epoch"] = epoch
                    target_interval = str(epoch).zfill(2)
                    for interval in session_intervals:
                        if (
                            target_interval in interval
                        ):  # TODO this is not true for the beans file
                            break
                    # TODO case when interval is not found is not handled
                    key["interval_list_name"] = interval
                    task_inserts.append(key.copy())
        self.insert(task_inserts, allow_direct_insert=True)

    @classmethod
    def update_entries(cls, restrict={}):
        existing_entries = (cls & restrict).fetch("KEY")
        for row in existing_entries:
            if (cls & row).fetch1("camera_names"):
                continue
            row["camera_names"] = [
                {"camera_name": (cls & row).fetch1("camera_name")}
            ]
            cls.update1(row=row)

    @classmethod
    def check_task_table(cls, task_table) -> bool:
        """Check whether the pynwb DynamicTable containing task metadata
        conforms to the expected format.

        The table should be an instance of pynwb.core.DynamicTable and contain
        the columns 'task_name', 'task_description', 'camera_id', 'and
        'task_epochs'.

        Parameters
        ----------
        task_table : pynwb.core.DynamicTable
            The table representing task metadata.

        Returns
        -------
        bool
            Whether the DynamicTable conforms to the expected format for
            loading data into the TaskEpoch table.
        """

        # TODO this could be more strict and check data types, but really it
        # should be schematized
        return (
            Task.check_task_table(task_table)
            and hasattr(task_table, "camera_id")
            and hasattr(task_table, "task_epochs")
        )

check_task_table(task_table) classmethod

Check whether the pynwb DynamicTable containing task metadata conforms to the expected format.

The table should be an instance of pynwb.core.DynamicTable and contain the columns 'task_name', 'task_description', 'camera_id', 'and 'task_epochs'.

Parameters:

Name Type Description Default
task_table DynamicTable

The table representing task metadata.

 required

Returns:

Type Description
bool

Whether the DynamicTable conforms to the expected format for loading data into the TaskEpoch table.
Source code in src/spyglass/common/common_task.py 
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
@classmethod
def check_task_table(cls, task_table) -> bool:
    """Check whether the pynwb DynamicTable containing task metadata
    conforms to the expected format.

    The table should be an instance of pynwb.core.DynamicTable and contain
    the columns 'task_name', 'task_description', 'camera_id', 'and
    'task_epochs'.

    Parameters
    ----------
    task_table : pynwb.core.DynamicTable
        The table representing task metadata.

    Returns
    -------
    bool
        Whether the DynamicTable conforms to the expected format for
        loading data into the TaskEpoch table.
    """

    # TODO this could be more strict and check data types, but really it
    # should be schematized
    return (
        Task.check_task_table(task_table)
        and hasattr(task_table, "camera_id")
        and hasattr(task_table, "task_epochs")
    )