Skip to content

common_behav.py

PositionSource

Bases: SpyglassIngestion, Manual

Source code in src/spyglass/common/common_behav.py 
 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
@schema
class PositionSource(SpyglassIngestion, dj.Manual):
    definition = """
    -> Session
    -> IntervalList
    ---
    source: varchar(200)             # source of data (e.g., trodes, dlc)
    import_file_name: varchar(2000)  # path to import file if importing
    """

    # Position intervals may already exist from a previous ingestion, so an
    # existing entry is validated rather than reinserted. NOTE: the previous
    # implementation used IntervalList.cautious_insert(update=True), which
    # silently overwrote a differing entry; validation prompts instead.
    _expected_duplicates = True

    class SpatialSeries(SpyglassIngestion, dj.Part):
        definition = """
        -> master
        id = 0 : int unsigned            # index of spatial series
        ---
        name=null: varchar(32)       # name of spatial series
        """

        _expected_duplicates = True  # follows the master

    _source_nwb_object_type = pynwb.behavior.Position

    # Entries are grouped by epoch in the override, not per column.
    table_key_to_obj_attr = {"self": dict()}

    def get_nwb_objects(self, nwb_file, nwb_file_name=None):
        """The file's Position interface holds every spatial series.

        `get_position_obj` finds series the default type filter would miss,
        and raises if a file declares more than one Position interface.
        """
        pos_interface = get_position_obj(nwb_file)
        return [pos_interface] if pos_interface is not None else []

    def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
        """Group the file's spatial series into one source entry per epoch.

        Each epoch yields an IntervalList entry (the parent, first), a source
        entry, and one SpatialSeries part entry per series in that epoch.
        RawPosition and its PosObject part are filled from the same series --
        they hold the object ids this pass already read -- rather than making
        RawPosition parse the file a second time.
        """
        sess_key = dict(base_key or dict())
        nwb_file_name = sess_key["nwb_file_name"]
        src_key = dict(**sess_key, source="imported", import_file_name="")

        all_pos = _get_pos_dict(
            position=nwb_obj.spatial_series,
            epoch_groups=_get_epoch_groups(nwb_obj),
            session_id=nwb_file_name,
            verbose=True,
        )
        if len(all_pos) == 0:
            self._info_msg(
                f"No position data found in {nwb_file_name}. Skipping."
            )
            return {self: []}

        intervals, sources, spat_series = [], [], []
        raw_pos, pos_objects = [], []

        for epoch, epoch_list in all_pos.items():
            ind_key = dict(interval_list_name=self.get_pos_interval_name(epoch))

            sources.append(dict(**src_key, **ind_key))
            raw_pos.append(dict(**sess_key, **ind_key))
            intervals.append(
                dict(
                    **sess_key,
                    **ind_key,
                    valid_times=epoch_list[0]["valid_times"],
                    pipeline="position",
                )
            )

            for index, pdict in enumerate(epoch_list):
                spat_series.append(
                    dict(
                        **sess_key,
                        **ind_key,
                        id=index,
                        name=pdict.get("name"),
                    )
                )
                pos_objects.append(
                    dict(
                        **sess_key,
                        **ind_key,
                        id=index,
                        raw_position_object_id=pdict["raw_position_object_id"],
                    )
                )

        return {
            IntervalList: intervals,
            self: sources,
            self.SpatialSeries: spat_series,
            RawPosition: raw_pos,
            RawPosition.PosObject: pos_objects,
        }

    def insert_from_nwbfile(self, nwb_file_name, config=None, dry_run=False):
        """Ingest, then map epoch intervals to the position intervals."""
        entries = super().insert_from_nwbfile(nwb_file_name, config, dry_run)

        if entries and not dry_run:
            populate_position_interval_map_session(nwb_file_name)

        return entries

    def make(self, keys: Union[List[Dict], dj.Table]):
        """Deprecated in favor of insert_from_nwbfile."""
        raise NotImplementedError(
            "PositionSource.make is deprecated. Use insert_from_nwbfile."
        )

    @staticmethod
    def get_pos_interval_name(epoch_num: int) -> str:
        """Return string of the interval name from the epoch number.

        Parameters
        ----------
        epoch_num : int
            Input epoch number

        Returns
        -------
        str
            Position interval name (e.g., pos 2 valid times)
        """
        try:
            int(epoch_num)
        except ValueError:
            raise ValueError(
                f"Epoch number must must be an integer. Received: {epoch_num}"
            )
        return f"pos {epoch_num} valid times"

    @staticmethod
    def _is_valid_name(name) -> bool:
        return name.startswith("pos ") and name.endswith(" valid times")

    @staticmethod
    def get_epoch_num(name: str) -> int:
        """Return the epoch number from the interval name.

        Parameters
        ----------
        name : str
            Name of position interval (e.g., pos epoch 1 index 2 valid times)

        Returns
        -------
        int
            epoch number
        """
        if not PositionSource._is_valid_name(name):
            raise ValueError(f"Invalid interval name: {name}")
        return int(name.replace("pos ", "").replace(" valid times", ""))

get_nwb_objects(nwb_file, nwb_file_name=None)

The file's Position interface holds every spatial series.

get_position_obj finds series the default type filter would miss, and raises if a file declares more than one Position interface.

Source code in src/spyglass/common/common_behav.py 
64
65
66
67
68
69
70
71
def get_nwb_objects(self, nwb_file, nwb_file_name=None):
    """The file's Position interface holds every spatial series.

    `get_position_obj` finds series the default type filter would miss,
    and raises if a file declares more than one Position interface.
    """
    pos_interface = get_position_obj(nwb_file)
    return [pos_interface] if pos_interface is not None else []

generate_entries_from_nwb_object(nwb_obj, base_key=None)

Group the file's spatial series into one source entry per epoch.

Each epoch yields an IntervalList entry (the parent, first), a source entry, and one SpatialSeries part entry per series in that epoch. RawPosition and its PosObject part are filled from the same series -- they hold the object ids this pass already read -- rather than making RawPosition parse the file a second time.

Source code in src/spyglass/common/common_behav.py 
 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
def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
    """Group the file's spatial series into one source entry per epoch.

    Each epoch yields an IntervalList entry (the parent, first), a source
    entry, and one SpatialSeries part entry per series in that epoch.
    RawPosition and its PosObject part are filled from the same series --
    they hold the object ids this pass already read -- rather than making
    RawPosition parse the file a second time.
    """
    sess_key = dict(base_key or dict())
    nwb_file_name = sess_key["nwb_file_name"]
    src_key = dict(**sess_key, source="imported", import_file_name="")

    all_pos = _get_pos_dict(
        position=nwb_obj.spatial_series,
        epoch_groups=_get_epoch_groups(nwb_obj),
        session_id=nwb_file_name,
        verbose=True,
    )
    if len(all_pos) == 0:
        self._info_msg(
            f"No position data found in {nwb_file_name}. Skipping."
        )
        return {self: []}

    intervals, sources, spat_series = [], [], []
    raw_pos, pos_objects = [], []

    for epoch, epoch_list in all_pos.items():
        ind_key = dict(interval_list_name=self.get_pos_interval_name(epoch))

        sources.append(dict(**src_key, **ind_key))
        raw_pos.append(dict(**sess_key, **ind_key))
        intervals.append(
            dict(
                **sess_key,
                **ind_key,
                valid_times=epoch_list[0]["valid_times"],
                pipeline="position",
            )
        )

        for index, pdict in enumerate(epoch_list):
            spat_series.append(
                dict(
                    **sess_key,
                    **ind_key,
                    id=index,
                    name=pdict.get("name"),
                )
            )
            pos_objects.append(
                dict(
                    **sess_key,
                    **ind_key,
                    id=index,
                    raw_position_object_id=pdict["raw_position_object_id"],
                )
            )

    return {
        IntervalList: intervals,
        self: sources,
        self.SpatialSeries: spat_series,
        RawPosition: raw_pos,
        RawPosition.PosObject: pos_objects,
    }

insert_from_nwbfile(nwb_file_name, config=None, dry_run=False)

Ingest, then map epoch intervals to the position intervals.

Source code in src/spyglass/common/common_behav.py 
141
142
143
144
145
146
147
148
def insert_from_nwbfile(self, nwb_file_name, config=None, dry_run=False):
    """Ingest, then map epoch intervals to the position intervals."""
    entries = super().insert_from_nwbfile(nwb_file_name, config, dry_run)

    if entries and not dry_run:
        populate_position_interval_map_session(nwb_file_name)

    return entries

make(keys)

Deprecated in favor of insert_from_nwbfile.

Source code in src/spyglass/common/common_behav.py 
150
151
152
153
154
def make(self, keys: Union[List[Dict], dj.Table]):
    """Deprecated in favor of insert_from_nwbfile."""
    raise NotImplementedError(
        "PositionSource.make is deprecated. Use insert_from_nwbfile."
    )

get_pos_interval_name(epoch_num) staticmethod

Return string of the interval name from the epoch number.

Parameters:

Name Type Description Default
epoch_num int

Input epoch number

 required

Returns:

Type Description
str

Position interval name (e.g., pos 2 valid times)
Source code in src/spyglass/common/common_behav.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@staticmethod
def get_pos_interval_name(epoch_num: int) -> str:
    """Return string of the interval name from the epoch number.

    Parameters
    ----------
    epoch_num : int
        Input epoch number

    Returns
    -------
    str
        Position interval name (e.g., pos 2 valid times)
    """
    try:
        int(epoch_num)
    except ValueError:
        raise ValueError(
            f"Epoch number must must be an integer. Received: {epoch_num}"
        )
    return f"pos {epoch_num} valid times"

get_epoch_num(name) staticmethod

Return the epoch number from the interval name.

Parameters:

Name Type Description Default
name str

Name of position interval (e.g., pos epoch 1 index 2 valid times)

 required

Returns:

Type Description
int

epoch number
Source code in src/spyglass/common/common_behav.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@staticmethod
def get_epoch_num(name: str) -> int:
    """Return the epoch number from the interval name.

    Parameters
    ----------
    name : str
        Name of position interval (e.g., pos epoch 1 index 2 valid times)

    Returns
    -------
    int
        epoch number
    """
    if not PositionSource._is_valid_name(name):
        raise ValueError(f"Invalid interval name: {name}")
    return int(name.replace("pos ", "").replace(" valid times", ""))

RawPosition

Bases: SpyglassIngestion, Imported

Source code in src/spyglass/common/common_behav.py 
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
239
240
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
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
@schema
class RawPosition(SpyglassIngestion, dj.Imported):
    definition = """
    -> PositionSource
    """

    # Filled by PositionSource, which reads the same spatial series.
    _expected_duplicates = True

    def insert_from_nwbfile(self, nwb_file_name, config=None, dry_run=False):
        """Defer to PositionSource, which generates this table's entries.

        The two tables describe the same spatial series, so parsing the file
        once fills both. Kept so `populate` and direct callers still work.
        """
        return PositionSource().insert_from_nwbfile(
            nwb_file_name, config, dry_run
        )

    class PosObject(SpyglassIngestion, dj.Part):
        definition = """
        -> master
        -> PositionSource.SpatialSeries.proj('id')
        ---
        raw_position_object_id: varchar(40) # id of spatial series in NWB file
        """

        _nwb_table = Nwbfile
        _expected_duplicates = True  # follows the master

        def fetch1_dataframe(self):
            """Return a dataframe with all RawPosition.PosObject items."""
            id_rp = [(n["id"], n["raw_position"]) for n in self.fetch_nwb()]

            if len(set(rp.interval for _, rp in id_rp)) > 1:
                logger.warning("Loading DataFrame with multiple intervals.")

            df_list = [
                pd.DataFrame(
                    data=rp.data,
                    index=pd.Index(rp.timestamps, name="time"),
                    columns=self._get_column_names(rp, pos_id),
                )
                for pos_id, rp in id_rp
            ]

            return reduce(lambda x, y: pd.merge(x, y, on="time"), df_list)

        @staticmethod
        def _get_column_names(rp, pos_id):
            INDEX_ADJUST = 1  # adjust 0-index to 1-index (e.g., xloc0 -> xloc1)
            n_pos_dims = rp.data.shape[1]
            column_names = [
                (
                    col  # use existing columns if already numbered
                    if "1" in rp.description or "2" in rp.description
                    # else number them by id
                    else col + str(pos_id + INDEX_ADJUST)
                )
                for col in rp.description.split(", ")
            ]
            if len(column_names) != n_pos_dims:
                # if the string split didn't work, use default names
                column_names = ["x", "y", "z"][:n_pos_dims]
            return column_names

    def make(self, key):
        """Deprecated in favor of insert_from_nwbfile."""
        raise NotImplementedError(
            "RawPosition.make is deprecated. Use insert_from_nwbfile."
        )

    def fetch_nwb(self, *attrs, **kwargs) -> list:
        """
        Returns a condatenated list of nwb objects from RawPosition.PosObject
        """
        return (
            self.PosObject()
            .restrict(self.restriction)  # Avoids fetch_nwb on whole table
            .fetch_nwb(*attrs, **kwargs)
        )

    def fetch1_dataframe(self):
        """Returns a dataframe with all RawPosition.PosObject items.

        Uses interval_list_name as column index.
        """
        ret = {}

        pos_obj_set = self.PosObject & self.restriction
        unique_intervals = set(pos_obj_set.fetch("interval_list_name"))

        for interval in unique_intervals:
            ret[interval] = (
                pos_obj_set & {"interval_list_name": interval}
            ).fetch1_dataframe()

        if len(unique_intervals) == 1:
            return next(iter(ret.values()))

        return pd.concat(ret, axis=1)

insert_from_nwbfile(nwb_file_name, config=None, dry_run=False)

Defer to PositionSource, which generates this table's entries.

The two tables describe the same spatial series, so parsing the file once fills both. Kept so populate and direct callers still work.

Source code in src/spyglass/common/common_behav.py 
210
211
212
213
214
215
216
217
218
def insert_from_nwbfile(self, nwb_file_name, config=None, dry_run=False):
    """Defer to PositionSource, which generates this table's entries.

    The two tables describe the same spatial series, so parsing the file
    once fills both. Kept so `populate` and direct callers still work.
    """
    return PositionSource().insert_from_nwbfile(
        nwb_file_name, config, dry_run
    )

PosObject

Bases: SpyglassIngestion, Part

Source code in src/spyglass/common/common_behav.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
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
class PosObject(SpyglassIngestion, dj.Part):
    definition = """
    -> master
    -> PositionSource.SpatialSeries.proj('id')
    ---
    raw_position_object_id: varchar(40) # id of spatial series in NWB file
    """

    _nwb_table = Nwbfile
    _expected_duplicates = True  # follows the master

    def fetch1_dataframe(self):
        """Return a dataframe with all RawPosition.PosObject items."""
        id_rp = [(n["id"], n["raw_position"]) for n in self.fetch_nwb()]

        if len(set(rp.interval for _, rp in id_rp)) > 1:
            logger.warning("Loading DataFrame with multiple intervals.")

        df_list = [
            pd.DataFrame(
                data=rp.data,
                index=pd.Index(rp.timestamps, name="time"),
                columns=self._get_column_names(rp, pos_id),
            )
            for pos_id, rp in id_rp
        ]

        return reduce(lambda x, y: pd.merge(x, y, on="time"), df_list)

    @staticmethod
    def _get_column_names(rp, pos_id):
        INDEX_ADJUST = 1  # adjust 0-index to 1-index (e.g., xloc0 -> xloc1)
        n_pos_dims = rp.data.shape[1]
        column_names = [
            (
                col  # use existing columns if already numbered
                if "1" in rp.description or "2" in rp.description
                # else number them by id
                else col + str(pos_id + INDEX_ADJUST)
            )
            for col in rp.description.split(", ")
        ]
        if len(column_names) != n_pos_dims:
            # if the string split didn't work, use default names
            column_names = ["x", "y", "z"][:n_pos_dims]
        return column_names

fetch1_dataframe()

Return a dataframe with all RawPosition.PosObject items.

Source code in src/spyglass/common/common_behav.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def fetch1_dataframe(self):
    """Return a dataframe with all RawPosition.PosObject items."""
    id_rp = [(n["id"], n["raw_position"]) for n in self.fetch_nwb()]

    if len(set(rp.interval for _, rp in id_rp)) > 1:
        logger.warning("Loading DataFrame with multiple intervals.")

    df_list = [
        pd.DataFrame(
            data=rp.data,
            index=pd.Index(rp.timestamps, name="time"),
            columns=self._get_column_names(rp, pos_id),
        )
        for pos_id, rp in id_rp
    ]

    return reduce(lambda x, y: pd.merge(x, y, on="time"), df_list)

make(key)

Deprecated in favor of insert_from_nwbfile.

Source code in src/spyglass/common/common_behav.py 
267
268
269
270
271
def make(self, key):
    """Deprecated in favor of insert_from_nwbfile."""
    raise NotImplementedError(
        "RawPosition.make is deprecated. Use insert_from_nwbfile."
    )

fetch_nwb(*attrs, **kwargs)

Returns a condatenated list of nwb objects from RawPosition.PosObject

Source code in src/spyglass/common/common_behav.py 
273
274
275
276
277
278
279
280
281
def fetch_nwb(self, *attrs, **kwargs) -> list:
    """
    Returns a condatenated list of nwb objects from RawPosition.PosObject
    """
    return (
        self.PosObject()
        .restrict(self.restriction)  # Avoids fetch_nwb on whole table
        .fetch_nwb(*attrs, **kwargs)
    )

fetch1_dataframe()

Returns a dataframe with all RawPosition.PosObject items.

Uses interval_list_name as column index.

Source code in src/spyglass/common/common_behav.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
def fetch1_dataframe(self):
    """Returns a dataframe with all RawPosition.PosObject items.

    Uses interval_list_name as column index.
    """
    ret = {}

    pos_obj_set = self.PosObject & self.restriction
    unique_intervals = set(pos_obj_set.fetch("interval_list_name"))

    for interval in unique_intervals:
        ret[interval] = (
            pos_obj_set & {"interval_list_name": interval}
        ).fetch1_dataframe()

    if len(unique_intervals) == 1:
        return next(iter(ret.values()))

    return pd.concat(ret, axis=1)

RawCompassDirection

Bases: SpyglassIngestion, Manual

Table to store raw CompassDirection data from NWB files.

Source code in src/spyglass/common/common_behav.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
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
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
@schema
class RawCompassDirection(SpyglassIngestion, dj.Manual):
    """
    Table to store raw CompassDirection data from NWB files.
    """

    definition = """
    -> Session
    -> IntervalList
    ---
    compass_object_id: varchar(40)  # the object id of the compass direction object
    name: varchar(80)              # name of the compass direction object
    """

    _nwb_table = Nwbfile
    _compass_import_enumerator = 1

    _source_nwb_object_type = CompassDirection

    @property
    def table_key_to_obj_attr(self):
        return {
            "self": {
                "name": "name",
                "compass_object_id": "object_id",
                "valid_times": self.generate_valid_intervals_from_timeseries,
                "interval_list_name": self.enumerated_interval_name,
            }
        }

    def get_nwb_objects(self, nwb_file, nwb_file_name=None):
        """Get all CompassDirection spatial series from NWB file, ordered by time."""
        compass_objects = super().get_nwb_objects(nwb_file, nwb_file_name)
        spatial_series = sum(
            [list(obj.spatial_series.values()) for obj in compass_objects], []
        )
        start_times = [ss.get_timestamps()[0] for ss in spatial_series]
        order = np.argsort(start_times)
        spatial_series = [spatial_series[i] for i in order]

        return spatial_series

    def enumerated_interval_name(
        self, obj: pynwb.behavior.SpatialSeries
    ) -> str:
        """Generate a unique interval list name for each compass direction object."""
        name = f"compass {self._compass_import_enumerator} valid times"
        self._compass_import_enumerator += 1
        return name

    @staticmethod
    def generate_valid_intervals_from_timeseries(
        nwb_obj: pynwb.behavior.SpatialSeries,
    ):
        """Generate valid intervals from spatial series.

        Parameters
        ----------
        nwb_obj : pynwb.behavior.SpatialSeries
            The pynwb.behavior.SpatialSeries NWB object.
        Returns
        -------
        valid_times : list
            List of valid time intervals.
        """
        timestamps = nwb_obj.get_timestamps()
        sampling_rate = estimate_sampling_rate(
            timestamps, filename=nwb_obj.name
        )
        valid_times = get_valid_intervals(
            timestamps=timestamps,
            sampling_rate=sampling_rate,
            min_valid_len=int(sampling_rate),
        )
        return valid_times

    def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
        """Add IntervalList entry to the generated entries."""
        super_ins = super().generate_entries_from_nwb_object(nwb_obj, base_key)
        self_key = super_ins[self][0]
        interval_insert = {
            k: v for k, v in self_key.items() if k in IntervalList.heading.names
        }
        self_key.pop(
            "valid_times", None
        )  # remove valid_times from the insert to this table
        return {
            IntervalList: [interval_insert],
            **super_ins,
        }

    def insert_from_nwbfile(self, nwb_file_name, config=None, dry_run=False):
        """Insert entries from NWB file, generating interval list names ordered by time."""
        self._compass_import_enumerator = 1  # reset enumerator
        return super().insert_from_nwbfile(nwb_file_name, config, dry_run)

get_nwb_objects(nwb_file, nwb_file_name=None)

Get all CompassDirection spatial series from NWB file, ordered by time.

Source code in src/spyglass/common/common_behav.py 
334
335
336
337
338
339
340
341
342
343
344
def get_nwb_objects(self, nwb_file, nwb_file_name=None):
    """Get all CompassDirection spatial series from NWB file, ordered by time."""
    compass_objects = super().get_nwb_objects(nwb_file, nwb_file_name)
    spatial_series = sum(
        [list(obj.spatial_series.values()) for obj in compass_objects], []
    )
    start_times = [ss.get_timestamps()[0] for ss in spatial_series]
    order = np.argsort(start_times)
    spatial_series = [spatial_series[i] for i in order]

    return spatial_series

enumerated_interval_name(obj)

Generate a unique interval list name for each compass direction object.

Source code in src/spyglass/common/common_behav.py 
346
347
348
349
350
351
352
def enumerated_interval_name(
    self, obj: pynwb.behavior.SpatialSeries
) -> str:
    """Generate a unique interval list name for each compass direction object."""
    name = f"compass {self._compass_import_enumerator} valid times"
    self._compass_import_enumerator += 1
    return name

generate_valid_intervals_from_timeseries(nwb_obj) staticmethod

Generate valid intervals from spatial series.

Parameters:

Name Type Description Default
nwb_obj SpatialSeries

The pynwb.behavior.SpatialSeries NWB object.

 required

Returns:

Name Type Description
valid_times list

List of valid time intervals.
Source code in src/spyglass/common/common_behav.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
@staticmethod
def generate_valid_intervals_from_timeseries(
    nwb_obj: pynwb.behavior.SpatialSeries,
):
    """Generate valid intervals from spatial series.

    Parameters
    ----------
    nwb_obj : pynwb.behavior.SpatialSeries
        The pynwb.behavior.SpatialSeries NWB object.
    Returns
    -------
    valid_times : list
        List of valid time intervals.
    """
    timestamps = nwb_obj.get_timestamps()
    sampling_rate = estimate_sampling_rate(
        timestamps, filename=nwb_obj.name
    )
    valid_times = get_valid_intervals(
        timestamps=timestamps,
        sampling_rate=sampling_rate,
        min_valid_len=int(sampling_rate),
    )
    return valid_times

generate_entries_from_nwb_object(nwb_obj, base_key=None)

Add IntervalList entry to the generated entries.

Source code in src/spyglass/common/common_behav.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
    """Add IntervalList entry to the generated entries."""
    super_ins = super().generate_entries_from_nwb_object(nwb_obj, base_key)
    self_key = super_ins[self][0]
    interval_insert = {
        k: v for k, v in self_key.items() if k in IntervalList.heading.names
    }
    self_key.pop(
        "valid_times", None
    )  # remove valid_times from the insert to this table
    return {
        IntervalList: [interval_insert],
        **super_ins,
    }

insert_from_nwbfile(nwb_file_name, config=None, dry_run=False)

Insert entries from NWB file, generating interval list names ordered by time.

Source code in src/spyglass/common/common_behav.py 
395
396
397
398
def insert_from_nwbfile(self, nwb_file_name, config=None, dry_run=False):
    """Insert entries from NWB file, generating interval list names ordered by time."""
    self._compass_import_enumerator = 1  # reset enumerator
    return super().insert_from_nwbfile(nwb_file_name, config, dry_run)

StateScriptFile

Bases: SpyglassIngestion, Imported

Source code in src/spyglass/common/common_behav.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
@schema
class StateScriptFile(SpyglassIngestion, dj.Imported):
    definition = """
    -> TaskEpoch
    ---
    file_object_id: varchar(40)  # the object id of the file object
    """

    _nwb_table = Nwbfile
    # Exact class name, as ndx_franklab_novela may not be importable
    _source_nwb_object_type = "AssociatedFiles"

    # An associated file is a state script if its description says so.
    # Matching ignores case and spaces, so "STATE SCRIPT" is covered too.
    _source_nwb_object_description = ("state script", "state_script")

    table_key_to_obj_attr = {"self": {"file_object_id": "object_id"}}

    def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
        """Expand one associated file into an entry per task epoch.

        The file names its epochs in a comma-separated string. Only epochs
        already present in TaskEpoch yield an entry, matching the previous
        implementation, which ran once per existing TaskEpoch key.
        """
        super_ins = super().generate_entries_from_nwb_object(nwb_obj, base_key)
        self_key = super_ins[self][0]

        # TODO: update associated_file_obj.task_epochs to be an array of
        # 1-based ints, not a comma-separated string of ints
        named_epochs = str(nwb_obj.task_epochs).split(",")
        task_epochs = TaskEpoch & {
            "nwb_file_name": self_key.get("nwb_file_name")
        }

        return {
            self: [
                dict(self_key, epoch=epoch)
                for epoch in task_epochs.fetch("epoch")
                if str(epoch) in named_epochs
            ]
        }

    def make(self, key):
        """Deprecated in favor of insert_from_nwbfile."""
        raise NotImplementedError(
            "StateScriptFile.make is deprecated. Use insert_from_nwbfile."
        )

generate_entries_from_nwb_object(nwb_obj, base_key=None)

Expand one associated file into an entry per task epoch.

The file names its epochs in a comma-separated string. Only epochs already present in TaskEpoch yield an entry, matching the previous implementation, which ran once per existing TaskEpoch key.

Source code in src/spyglass/common/common_behav.py 
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
    """Expand one associated file into an entry per task epoch.

    The file names its epochs in a comma-separated string. Only epochs
    already present in TaskEpoch yield an entry, matching the previous
    implementation, which ran once per existing TaskEpoch key.
    """
    super_ins = super().generate_entries_from_nwb_object(nwb_obj, base_key)
    self_key = super_ins[self][0]

    # TODO: update associated_file_obj.task_epochs to be an array of
    # 1-based ints, not a comma-separated string of ints
    named_epochs = str(nwb_obj.task_epochs).split(",")
    task_epochs = TaskEpoch & {
        "nwb_file_name": self_key.get("nwb_file_name")
    }

    return {
        self: [
            dict(self_key, epoch=epoch)
            for epoch in task_epochs.fetch("epoch")
            if str(epoch) in named_epochs
        ]
    }

make(key)

Deprecated in favor of insert_from_nwbfile.

Source code in src/spyglass/common/common_behav.py 
444
445
446
447
448
def make(self, key):
    """Deprecated in favor of insert_from_nwbfile."""
    raise NotImplementedError(
        "StateScriptFile.make is deprecated. Use insert_from_nwbfile."
    )

VideoFile

Bases: SpyglassIngestion, Imported

Video file metadata from NWB ImageSeries.

Notes

The video timestamps come from: videoTimeStamps.cameraHWSync if PTP is used. If PTP is not used, the video timestamps come from videoTimeStamps.cameraHWFrameCount.

Issue #1444 Note: VideoFile requires TaskEpoch entries to import videos. If your NWB file contains ImageSeries (video data) without task metadata, a warning will be issued.

Source code in src/spyglass/common/common_behav.py 
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
@schema
class VideoFile(SpyglassIngestion, dj.Imported):
    """Video file metadata from NWB ImageSeries.

    Notes
    -----
    The video timestamps come from: videoTimeStamps.cameraHWSync if PTP is
    used. If PTP is not used, the video timestamps come from
    videoTimeStamps.cameraHWFrameCount.

    **Issue #1444 Note:** VideoFile requires TaskEpoch entries to import videos.
    If your NWB file contains ImageSeries (video data) without task metadata,
    a warning will be issued.
    """

    definition = """
    -> TaskEpoch
    video_file_num = 0: int
    ---
    camera_name: varchar(80)
    video_file_object_id: varchar(40)  # the object id of the file object
    """

    _nwb_table = Nwbfile
    _timestamp_overlap_threshold = 0.9  # Min fraction of timestamps in epoch
    _epoch_cache = dict()  # nwb_file_name -> {epoch: valid times}
    _failed_videos = defaultdict(list)  # reset per ingested file
    _video_count = 0  # ImageSeries seen in the file being ingested
    _placed_videos = 0  # ImageSeries that landed in at least one epoch

    _source_nwb_object_type = pynwb.image.ImageSeries

    # Entries are built per epoch in the override, not per column.
    table_key_to_obj_attr = {"self": dict()}

    def _epoch_intervals(self, nwb_file_name) -> dict:
        """Return the valid times of each task epoch, fetched once per file.

        A video belongs to whichever epoch its timestamps fall inside, so
        every epoch's times are needed to place a single video.

        Parameters
        ----------
        nwb_file_name : str
            The file being ingested.

        Returns
        -------
        dict
            Epoch number to that epoch's Interval.
        """
        if nwb_file_name not in self._epoch_cache:
            self._epoch_cache[nwb_file_name] = {
                epoch: (
                    IntervalList
                    & {
                        "nwb_file_name": nwb_file_name,
                        "interval_list_name": interval_list_name,
                    }
                ).fetch_interval()
                for epoch, interval_list_name in zip(
                    *(TaskEpoch & {"nwb_file_name": nwb_file_name}).fetch(
                        "epoch", "interval_list_name"
                    )
                )
            }
        return self._epoch_cache[nwb_file_name]

    def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
        """Place one video in whichever epochs its timestamps overlap.

        Failures are collected rather than raised, matching the previous
        implementation, which reported them per file at the end.
        """
        base_key = base_key or dict()
        self._video_count += 1
        entries = []
        mismatches = []  # kept only if the video places in no epoch

        for epoch, valid_times in self._epoch_intervals(
            base_key["nwb_file_name"]
        ).items():
            key = dict(base_key, epoch=epoch)
            try:
                rows, failure_reason, overlap_percent = (
                    self._validate_video_timestamps(nwb_obj, valid_times, key)
                )
            except KeyError as err:  # camera device not in CameraDevice
                self._failed_videos["missing_camera"].append(
                    {
                        "name": nwb_obj.name,
                        "camera": getattr(
                            nwb_obj.device, "camera_name", "unknown"
                        ),
                        "error": str(err),
                    }
                )
                break  # the camera is missing for every epoch
            except Exception as err:
                self._failed_videos["other"].append(
                    {
                        "name": nwb_obj.name,
                        "error": f"{type(err).__name__}: {str(err)}",
                    }
                )
                break

            if failure_reason:
                mismatches.append(
                    {
                        "name": nwb_obj.name,
                        "reason": failure_reason,
                        "overlap_percent": overlap_percent,
                    }
                )
            else:
                entries.extend(rows)

        # Counted per source series, not per row: one video spanning several
        # epochs yields several rows, so a row count cannot tell whether
        # every series was placed.
        self._placed_videos += bool(entries)

        # Every epoch a video does *not* belong to fails the overlap check, so
        # a placed video would otherwise report a mismatch for each of its
        # non-owning epochs. Only a video that landed nowhere has failed.
        if not entries:
            self._failed_videos["timestamp_mismatch"].extend(mismatches)

        return {self: entries}

    def insert_from_nwbfile(self, nwb_file_name, config=None, dry_run=False):
        """Ingest, then report on any videos that could not be placed."""
        self._epoch_cache = dict()
        self._failed_videos = defaultdict(list)
        self._video_count = 0
        self._placed_videos = 0

        entries = super().insert_from_nwbfile(nwb_file_name, config, dry_run)

        if not self._video_count:
            self._warn_msg(
                f"No video data interface found in {nwb_file_name}\n"
            )
            return entries

        if self._placed_videos < self._video_count:
            self._report_partial_import(
                nwb_file_name,
                self._failed_videos,
                self._video_count,
                self._placed_videos,
            )

        return entries

    def _prepare_video_entry(
        self, key, video_obj, cam_device_regex: str = r"camera_device (\d+)"
    ):
        """Prepare a VideoFile entry dict for a given video object.

        Parameters
        ----------
        key : dict
            The primary key for the VideoFile entry
        video_obj : pynwb.image.ImageSeries
            The video object from the NWB file
        cam_device_regex : str, optional
            Regular expression pattern to extract camera device number.
            Default: r"camera_device (\\d+)"

        Returns
        -------
        dict
            Prepared entry dict ready for insertion

        Raises
        ------
        KeyError
            If camera_name is not found in CameraDevice table
        """
        nwb_cam_device = video_obj.device.name
        camera_name = video_obj.device.camera_name

        if not (CameraDevice & {"camera_name": camera_name}):
            raise KeyError(
                f"No camera with camera_name: {camera_name} found "
                "in CameraDevice table."
            )

        match = re.match(cam_device_regex, nwb_cam_device)
        if not match:
            raise ValueError(
                f"Camera device name '{nwb_cam_device}' does not match "
                f"expected pattern '{cam_device_regex}'"
            )

        return dict(
            key,
            video_file_num=int(match[1]),
            camera_name=camera_name,
            video_file_object_id=video_obj.object_id,
        )

    def _validate_video_timestamps(self, video_obj, valid_times, key):
        """Validate video timestamps and return entries or failure reason.

        Handles both single-file and multi-file ImageSeries. Validates that
        timestamps meet the overlap threshold with epoch intervals.

        Parameters
        ----------
        video_obj : pynwb.image.ImageSeries
            The video object from the NWB file
        valid_times : Interval
            Valid time intervals for the current epoch
        key : dict
            The primary key for the VideoFile entry

        Returns
        -------
        tuple
            (entries_list, failure_reason_or_None, overlap_percent)
            - If validation passes: ([entry_dicts], None, overlap_percent)
            - If validation fails: ([], "failure reason string", overlap_percent)
        """
        timestamps = video_obj.timestamps
        starting_frame = getattr(video_obj, "starting_frame", None)

        # Multi-file ImageSeries
        if starting_frame is not None and len(starting_frame) > 1:
            entries, overlap_pct = self._validate_multifile_timestamps(
                video_obj, timestamps, starting_frame, valid_times, key
            )
            if not entries:
                threshold_pct = self._timestamp_overlap_threshold * 100
                return (
                    [],
                    (
                        f"No file segments have ≥{threshold_pct:.0f}% "
                        "timestamp overlap with epoch"
                    ),
                    overlap_pct,
                )
            return entries, None, overlap_pct

        # Single-file ImageSeries: valid if >= threshold% of timestamps
        # overlap with epoch intervals (epoch covers the video).
        these_times = valid_times.contains(timestamps)
        overlap_pct = len(these_times) / len(timestamps)

        # Also valid if a single epoch interval is >= threshold% covered
        # by the video timestamps (video covers the whole epoch).
        timestamps_interval = [timestamps[0], timestamps[-1]]
        max_interval_overlap_pct = 0
        for interval in valid_times.times:
            interval_duration = interval[1] - interval[0]
            if interval_duration <= 0:
                continue
            overlap_start = max(interval[0], timestamps_interval[0])
            overlap_end = min(interval[1], timestamps_interval[1])
            overlap_duration = max(0, overlap_end - overlap_start)
            interval_overlap_pct = overlap_duration / interval_duration
            max_interval_overlap_pct = max(
                max_interval_overlap_pct, interval_overlap_pct
            )

        if (
            overlap_pct < self._timestamp_overlap_threshold
            and max_interval_overlap_pct < self._timestamp_overlap_threshold
        ):
            threshold_pct = self._timestamp_overlap_threshold * 100
            return (
                [],
                (
                    f"Only {overlap_pct:.1%} of timestamps overlap with epoch, "
                    f"and the best-covered epoch interval has only "
                    f"{max_interval_overlap_pct:.1%} of its duration covered "
                    f"by the timestamps (need ≥{threshold_pct:.0f}%)"
                ),
                overlap_pct,
            )

        return [self._prepare_video_entry(key, video_obj)], None, overlap_pct

    def _validate_multifile_timestamps(
        self,
        video_obj,
        timestamps,
        starting_frame,
        valid_times,
        key,
    ):
        """Validate each segment of multi-file ImageSeries timestamps.

        Parameters
        ----------
        video_obj : pynwb.image.ImageSeries
            The video object from the NWB file
        timestamps : array
            All timestamps for the ImageSeries
        starting_frame : array
            Frame indices indicating where each external file begins
        valid_times : Interval
            Valid time intervals for the current epoch
        key : dict
            The primary key for the VideoFile entry

        Returns
        -------
        tuple(list, float)
            List of entry dicts for segments with valid timestamps. May be empty
            Maximum overlap percentage across all file segments
        """
        entries = []

        max_overlap_pct = 0
        for file_idx in range(len(starting_frame)):
            # Determine timestamp range for this file segment
            start_idx = starting_frame[file_idx]
            end_idx = (
                starting_frame[file_idx + 1]
                if file_idx + 1 < len(starting_frame)
                else len(timestamps)
            )

            # Extract timestamps for this specific file
            file_timestamps = timestamps[start_idx:end_idx]

            # Check if threshold % of this file's timestamps overlap with epoch
            these_times = valid_times.contains(file_timestamps)
            overlap_pct = len(these_times) / len(file_timestamps)
            max_overlap_pct = max(max_overlap_pct, overlap_pct)

            if len(these_times) < (
                self._timestamp_overlap_threshold * len(file_timestamps)
            ):
                continue

            # This file segment matches the epoch - prepare VideoFile entry
            entry = self._prepare_video_entry(key.copy(), video_obj)
            entries.append(entry)

        return entries, max_overlap_pct

    def make(self, key, verbose=True, skip_duplicates=False):
        """Deprecated in favor of insert_from_nwbfile."""
        raise NotImplementedError(
            "VideoFile.make is deprecated. Use insert_from_nwbfile."
        )

    @staticmethod
    def _report_partial_import(
        nwb_file_name, failed_videos, total_videos, imported_count
    ):
        """Report specific reasons for partial video import.

        Issue #1444: Provide detailed diagnostics for each video that wasn't
        imported, categorized by failure reason with specific details.

        Parameters
        ----------
        nwb_file_name : str
            Name of the NWB file
        failed_videos : dict
            Dictionary with keys 'timestamp_mismatch', 'missing_camera',
            'other', each containing list of failure details
        total_videos : int
            Total number of ImageSeries found
        imported_count : int
            Number of ImageSeries successfully imported
        """

        msg_parts = [
            f"{nwb_file_name}: VideoFile Partial Import",
            f"Imported {imported_count}/{total_videos} ImageSeries",
        ]

        if failed_videos["timestamp_mismatch"]:
            msg_parts.append("\nTimestamp mismatches:")
            for item in failed_videos["timestamp_mismatch"]:
                msg_parts.append(f"  - {item['name']}: {item['reason']}")

        if failed_videos["missing_camera"]:
            msg_parts.append("\nMissing camera devices:")
            for item in failed_videos["missing_camera"]:
                msg_parts.append(
                    f"  - {item['name']}: camera '{item['camera']}' "
                    "not in CameraDevice table"
                )

        if failed_videos["other"]:
            msg_parts.append("\nOther errors:")
            for item in failed_videos["other"]:
                msg_parts.append(f"  - {item['name']}: {item['error']}")

        logger.warning("\n".join(msg_parts))

    @classmethod
    def update_entries(cls, restrict=True):
        """Update the camera_name field for all entries in the table."""
        existing_entries = (cls & restrict).fetch("KEY")
        for row in existing_entries:
            if (cls & row).fetch1("camera_name"):
                continue
            video_nwb = (cls & row).fetch_nwb()[0]
            if len(video_nwb) != 1:
                raise ValueError(
                    f"Expecting 1 video file per entry. {len(video_nwb)} found"
                )
            row["camera_name"] = video_nwb[0]["video_file"].device.camera_name
            cls.update1(row=row)

    @classmethod
    def get_abs_path(cls, key: Dict):
        """Return the absolute path for a stored video file given a key.

        Key must include the nwb_file_name and epoch number. The
        SPYGLASS_VIDEO_DIR environment variable must be set.

        Parameters
        ----------
        key : dict
            dictionary with nwb_file_name and epoch as keys

        Returns
        -------
        nwb_video_file_abspath : str
            The absolute path for the given file name.
        """
        video_path_obj = pathlib.Path(video_dir)
        video_info = (cls & key).fetch1()
        nwb_path = Nwbfile.get_abs_path(key["nwb_file_name"])
        nwbf = get_nwb_file(nwb_path)
        nwb_video = nwbf.objects[video_info["video_file_object_id"]]
        video_filename = nwb_video.name
        # see if the file exists and is stored in the base analysis dir
        nwb_video_file_abspath = pathlib.Path(video_path_obj / video_filename)
        if nwb_video_file_abspath.exists():
            return nwb_video_file_abspath.as_posix()
        else:
            raise FileNotFoundError(
                f"video file with filename: {video_filename} "
                f"does not exist in {video_path_obj}/"
            )

generate_entries_from_nwb_object(nwb_obj, base_key=None)

Place one video in whichever epochs its timestamps overlap.

Failures are collected rather than raised, matching the previous implementation, which reported them per file at the end.

Source code in src/spyglass/common/common_behav.py 
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
    """Place one video in whichever epochs its timestamps overlap.

    Failures are collected rather than raised, matching the previous
    implementation, which reported them per file at the end.
    """
    base_key = base_key or dict()
    self._video_count += 1
    entries = []
    mismatches = []  # kept only if the video places in no epoch

    for epoch, valid_times in self._epoch_intervals(
        base_key["nwb_file_name"]
    ).items():
        key = dict(base_key, epoch=epoch)
        try:
            rows, failure_reason, overlap_percent = (
                self._validate_video_timestamps(nwb_obj, valid_times, key)
            )
        except KeyError as err:  # camera device not in CameraDevice
            self._failed_videos["missing_camera"].append(
                {
                    "name": nwb_obj.name,
                    "camera": getattr(
                        nwb_obj.device, "camera_name", "unknown"
                    ),
                    "error": str(err),
                }
            )
            break  # the camera is missing for every epoch
        except Exception as err:
            self._failed_videos["other"].append(
                {
                    "name": nwb_obj.name,
                    "error": f"{type(err).__name__}: {str(err)}",
                }
            )
            break

        if failure_reason:
            mismatches.append(
                {
                    "name": nwb_obj.name,
                    "reason": failure_reason,
                    "overlap_percent": overlap_percent,
                }
            )
        else:
            entries.extend(rows)

    # Counted per source series, not per row: one video spanning several
    # epochs yields several rows, so a row count cannot tell whether
    # every series was placed.
    self._placed_videos += bool(entries)

    # Every epoch a video does *not* belong to fails the overlap check, so
    # a placed video would otherwise report a mismatch for each of its
    # non-owning epochs. Only a video that landed nowhere has failed.
    if not entries:
        self._failed_videos["timestamp_mismatch"].extend(mismatches)

    return {self: entries}

insert_from_nwbfile(nwb_file_name, config=None, dry_run=False)

Ingest, then report on any videos that could not be placed.

Source code in src/spyglass/common/common_behav.py 
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
def insert_from_nwbfile(self, nwb_file_name, config=None, dry_run=False):
    """Ingest, then report on any videos that could not be placed."""
    self._epoch_cache = dict()
    self._failed_videos = defaultdict(list)
    self._video_count = 0
    self._placed_videos = 0

    entries = super().insert_from_nwbfile(nwb_file_name, config, dry_run)

    if not self._video_count:
        self._warn_msg(
            f"No video data interface found in {nwb_file_name}\n"
        )
        return entries

    if self._placed_videos < self._video_count:
        self._report_partial_import(
            nwb_file_name,
            self._failed_videos,
            self._video_count,
            self._placed_videos,
        )

    return entries

make(key, verbose=True, skip_duplicates=False)

Deprecated in favor of insert_from_nwbfile.

Source code in src/spyglass/common/common_behav.py 
796
797
798
799
800
def make(self, key, verbose=True, skip_duplicates=False):
    """Deprecated in favor of insert_from_nwbfile."""
    raise NotImplementedError(
        "VideoFile.make is deprecated. Use insert_from_nwbfile."
    )

update_entries(restrict=True) classmethod

Update the camera_name field for all entries in the table.

Source code in src/spyglass/common/common_behav.py 
849
850
851
852
853
854
855
856
857
858
859
860
861
862
@classmethod
def update_entries(cls, restrict=True):
    """Update the camera_name field for all entries in the table."""
    existing_entries = (cls & restrict).fetch("KEY")
    for row in existing_entries:
        if (cls & row).fetch1("camera_name"):
            continue
        video_nwb = (cls & row).fetch_nwb()[0]
        if len(video_nwb) != 1:
            raise ValueError(
                f"Expecting 1 video file per entry. {len(video_nwb)} found"
            )
        row["camera_name"] = video_nwb[0]["video_file"].device.camera_name
        cls.update1(row=row)

get_abs_path(key) classmethod

Return the absolute path for a stored video file given a key.

Key must include the nwb_file_name and epoch number. The SPYGLASS_VIDEO_DIR environment variable must be set.

Parameters:

Name Type Description Default
key dict

dictionary with nwb_file_name and epoch as keys

 required

Returns:

Name Type Description
nwb_video_file_abspath str

The absolute path for the given file name.
Source code in src/spyglass/common/common_behav.py 
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
@classmethod
def get_abs_path(cls, key: Dict):
    """Return the absolute path for a stored video file given a key.

    Key must include the nwb_file_name and epoch number. The
    SPYGLASS_VIDEO_DIR environment variable must be set.

    Parameters
    ----------
    key : dict
        dictionary with nwb_file_name and epoch as keys

    Returns
    -------
    nwb_video_file_abspath : str
        The absolute path for the given file name.
    """
    video_path_obj = pathlib.Path(video_dir)
    video_info = (cls & key).fetch1()
    nwb_path = Nwbfile.get_abs_path(key["nwb_file_name"])
    nwbf = get_nwb_file(nwb_path)
    nwb_video = nwbf.objects[video_info["video_file_object_id"]]
    video_filename = nwb_video.name
    # see if the file exists and is stored in the base analysis dir
    nwb_video_file_abspath = pathlib.Path(video_path_obj / video_filename)
    if nwb_video_file_abspath.exists():
        return nwb_video_file_abspath.as_posix()
    else:
        raise FileNotFoundError(
            f"video file with filename: {video_filename} "
            f"does not exist in {video_path_obj}/"
        )

PositionIntervalMap

Bases: SpyglassMixin, Computed

Source code in src/spyglass/common/common_behav.py 
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
@schema
class PositionIntervalMap(SpyglassMixin, dj.Computed):
    definition = """
    -> IntervalList
    ---
    position_interval_name="": varchar(200)  # corresponding interval name
    """

    # #849 - Insert null to avoid rerun

    def make(self, key):
        """Make without transaction"""
        self._no_transaction_make(key)

    def _no_transaction_make(self, key):
        # Find correspondence between pos valid times names and epochs. Use
        # epsilon to tolerate small differences in epoch boundaries across
        # epoch/pos intervals

        if not self.connection.in_transaction:
            # if called w/o transaction, call add via `populate`
            self.populate(key)
            return
        if self & key:
            return

        # *** HARD CODED VALUES ***
        EPSILON = 0.51  # tolerated time diff in bounds across epoch/pos
        no_pop_msg = "CANNOT POPULATE PositionIntervalMap"

        # Strip extra info from key if not passed via populate call
        key = {k: v for k, v in key.items() if k in self.primary_key}

        nwb_file_name = key["nwb_file_name"]
        pos_intervals = get_pos_interval_list_names(nwb_file_name)
        null_key = dict(key, position_interval_name="")
        insert_opts = dict(allow_direct_insert=True, skip_duplicates=True)

        # Skip populating if no pos interval list names
        if len(pos_intervals) == 0:
            self._err_msg(f"NO POS INTERVALS FOR {key};\n{no_pop_msg}")
            self.insert1(null_key, **insert_opts)
            return

        valid_times = (IntervalList & key).fetch1("valid_times")
        time_bounds = [
            valid_times[0][0] - EPSILON,
            valid_times[-1][-1] + EPSILON,
        ]

        matching_pos_intervals = []
        restr = (
            f"nwb_file_name='{nwb_file_name}' AND interval_list_name=" + "'{}'"
        )
        for pos_interval in pos_intervals:
            pos_times = (IntervalList & restr.format(pos_interval)).fetch(
                "valid_times"
            )

            if len(pos_times) == 0:
                continue

            pos_times = pos_times[0]

            if all(
                [
                    time_bounds[0] <= time <= time_bounds[1]
                    for time in [pos_times[0][0], pos_times[-1][-1]]
                ]
            ):
                matching_pos_intervals.append(pos_interval)

            if len(matching_pos_intervals) > 1:
                break

        # Check that each pos interval was matched to only one epoch
        if len(matching_pos_intervals) != 1:
            self._warn_msg(
                f"{no_pop_msg}. Found {len(matching_pos_intervals)} pos "
                + f"intervals for\n\t{key}\n\t"
                + f"Matching intervals: {matching_pos_intervals}"
            )
            self.insert1(null_key, **insert_opts)
            return

        # Insert into table
        self.insert1(
            dict(key, position_interval_name=matching_pos_intervals[0]),
            **insert_opts,
        )
        self._info_msg(
            "Populated PosIntervalMap for "
            + f'{nwb_file_name}, {key["interval_list_name"]}'
        )

make(key)

Make without transaction

Source code in src/spyglass/common/common_behav.py 
908
909
910
def make(self, key):
    """Make without transaction"""
    self._no_transaction_make(key)

get_pos_interval_list_names(nwb_file_name)

Return a list of position interval list names for a given NWB file.

Source code in src/spyglass/common/common_behav.py 
 994
 995
 996
 997
 998
 999
1000
1001
1002
def get_pos_interval_list_names(nwb_file_name) -> list:
    """Return a list of position interval list names for a given NWB file."""
    return [
        interval_list_name
        for interval_list_name in (
            IntervalList & {"nwb_file_name": nwb_file_name}
        ).fetch("interval_list_name")
        if PositionSource._is_valid_name(interval_list_name)
    ]

convert_epoch_interval_name_to_position_interval_name(key, populate_missing=True)

Converts IntervalList key to the corresponding position interval name.

Parameters:

Name Type Description Default
key dict

Lookup key

 required
populate_missing bool

Whether to populate PositionIntervalMap for the key if missing. Should be False if this function is used inside of another populate call. Defaults to True

 True

Returns:

Name Type Description
position_interval_name str
Source code in src/spyglass/common/common_behav.py 
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
def convert_epoch_interval_name_to_position_interval_name(
    key: dict, populate_missing: bool = True
) -> str:
    """Converts IntervalList key to the corresponding position interval name.

    Parameters
    ----------
    key : dict
        Lookup key
    populate_missing: bool
        Whether to populate PositionIntervalMap for the key if missing. Should
        be False if this function is used inside of another populate call.
        Defaults to True

    Returns
    -------
    position_interval_name : str
    """
    # get the interval list name if given epoch but not interval list name
    if "interval_list_name" not in key and "epoch" in key:
        key["interval_list_name"] = get_interval_list_name_from_epoch(
            key["nwb_file_name"], key["epoch"]
        )

    pos_query = PositionIntervalMap & key
    pos_str = "position_interval_name"

    no_entries = len(pos_query) == 0
    null_entry = pos_query.fetch(pos_str)[0] == "" if len(pos_query) else False

    if populate_missing and (no_entries or null_entry):
        if null_entry:
            pos_query.delete(
                force_permission=True, safemode=False
            )  # no prompt; bypass delete permission check for null placeholder entry
        PositionIntervalMap()._no_transaction_make(key)
        pos_query = PositionIntervalMap & key

    if pos_query.fetch(pos_str)[0] == "":
        logger.info(f"No position intervals found for {key}")
        return []

    if len(pos_query) == 1:
        return pos_query.fetch1("position_interval_name")

    else:
        raise ValueError(f"Multiple intervals found for {key}: {pos_query}")

get_interval_list_name_from_epoch(nwb_file_name, epoch)

Returns the interval list name for the given epoch.

Parameters:

Name Type Description Default
nwb_file_name str

The name of the NWB file.

 required
epoch int

The epoch number.

 required

Returns:

Name Type Description
interval_list_name str

The interval list name.
Source code in src/spyglass/common/common_behav.py 
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
def get_interval_list_name_from_epoch(nwb_file_name: str, epoch: int) -> str:
    """Returns the interval list name for the given epoch.

    Parameters
    ----------
    nwb_file_name : str
        The name of the NWB file.
    epoch : int
        The epoch number.

    Returns
    -------
    interval_list_name : str
        The interval list name.
    """
    interval_names = (
        TaskEpoch & {"nwb_file_name": nwb_file_name, "epoch": epoch}
    ).fetch("interval_list_name")

    if len(interval_names) != 1:
        logger.info(
            f"Found {len(interval_names)} interval list names found for "
            + f"{nwb_file_name} epoch {epoch}"
        )
        return None

    return interval_names[0]

get_position_interval_epoch(nwb_file_name, position_interval_name)

Return the epoch number for a given position interval name.

Source code in src/spyglass/common/common_behav.py 
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
def get_position_interval_epoch(
    nwb_file_name: str, position_interval_name: str
) -> int:
    """Return the epoch number for a given position interval name."""
    # look up the epoch
    key = dict(
        nwb_file_name=nwb_file_name,
        position_interval_name=position_interval_name,
    )
    query = PositionIntervalMap * TaskEpoch & key
    if query:
        return query.fetch1("epoch")
    # if no match, make sure all epoch interval names are mapped
    for epoch_key in (TaskEpoch() & key).fetch(
        "nwb_file_name", "interval_list_name", as_dict=True
    ):
        convert_epoch_interval_name_to_position_interval_name(epoch_key)
    # try again
    query = PositionIntervalMap * TaskEpoch & key
    if query:
        return query.fetch1("epoch")
    return None

populate_position_interval_map_session(nwb_file_name)

Populate PositionIntervalMap for all epochs in a given NWB file.

Source code in src/spyglass/common/common_behav.py 
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
def populate_position_interval_map_session(nwb_file_name: str):
    """Populate PositionIntervalMap for all epochs in a given NWB file."""
    # 1. remove redundancy in interval names
    # 2. let PositionIntervalMap handle transaction context
    nwb_dict = dict(nwb_file_name=nwb_file_name)
    intervals = (TaskEpoch & nwb_dict).fetch("interval_list_name")
    for interval_name in set(intervals):
        interval_dict = dict(interval_list_name=interval_name)
        if PositionIntervalMap & interval_dict:
            continue
        PositionIntervalMap().make(dict(nwb_dict, **interval_dict))