Skip to content

common_interval.py

IntervalList

Bases: SpyglassMixin, Manual

Source code in src/spyglass/common/common_interval.py 
 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
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
@schema
class IntervalList(SpyglassMixin, dj.Manual):
    definition = """
    # Time intervals used for analysis
    -> Session
    interval_list_name: varchar(170)  # descriptive name of this interval list
    ---
    valid_times: longblob  # numpy array with start/end times for each interval
    pipeline = "": varchar(64)  # type of interval list
    """

    # See #630, #664. Excessive key length.

    @classmethod
    def insert_from_nwbfile(cls, nwbf: NWBFile, *, nwb_file_name: str):
        """Add each entry in the NWB file epochs table to the IntervalList.

        The interval list name for each epoch is set to the first tag for the
        epoch. If the epoch has no tags, then 'interval_x' will be used as the
        interval list name, where x is the index (0-indexed) of the epoch in the
        epochs table. The start time and stop time of the epoch are stored in
        the valid_times field as a numpy array of [start time, stop time] for
        each epoch.

        Parameters
        ----------
        nwbf : pynwb.NWBFile
            The source NWB file object.
        nwb_file_name : str
            The file name of the NWB file, used as a primary key to the Session
            table.
        """
        if nwbf.epochs is None:
            logger.info("No epochs found in NWB file.")
            return

        epochs = nwbf.epochs.to_dataframe()

        # Create a list of dictionaries to insert
        epoch_inserts = epochs.apply(
            lambda epoch_data: {
                "nwb_file_name": nwb_file_name,
                "interval_list_name": (
                    epoch_data.tags[0]
                    if epoch_data.tags
                    else f"interval_{epoch_data.name}"
                ),
                "valid_times": np.asarray(
                    [[epoch_data.start_time, epoch_data.stop_time]]
                ),
            },
            axis=1,
        ).tolist()

        cls.insert(epoch_inserts, skip_duplicates=True)

    def plot_intervals(self, figsize=(20, 5), return_fig=False):
        """Plot the intervals in the interval list."""
        interval_list = pd.DataFrame(self)
        fig, ax = plt.subplots(figsize=figsize)
        interval_count = 0
        for row in interval_list.itertuples(index=False):
            for interval in row.valid_times:
                ax.plot(interval, [interval_count, interval_count])
                ax.scatter(
                    interval,
                    [interval_count, interval_count],
                    alpha=0.8,
                    zorder=2,
                )
            interval_count += 1
        ax.set_yticks(np.arange(interval_list.shape[0]))
        ax.set_yticklabels(interval_list.interval_list_name)
        ax.set_xlabel("Time [s]")
        ax.grid(True)
        if return_fig:
            return fig

    def plot_epoch_pos_raw_intervals(self, figsize=(20, 5), return_fig=False):
        """Plot an epoch's position, raw data, and valid times intervals."""
        interval_list = pd.DataFrame(self)
        fig, ax = plt.subplots(figsize=(30, 3))

        raw_data_valid_times = interval_list.loc[
            interval_list.interval_list_name == "raw data valid times"
        ].valid_times
        interval_y = 1

        for interval in np.asarray(raw_data_valid_times)[0]:
            ax.plot(interval, [interval_y, interval_y])
            ax.scatter(interval, [interval_y, interval_y], alpha=0.8, zorder=2)

        epoch_valid_times = (
            interval_list.set_index("interval_list_name")
            .filter(regex=r"^[0-9]", axis=0)
            .valid_times
        )
        interval_y = 2
        for epoch, valid_times in zip(
            epoch_valid_times.index, epoch_valid_times
        ):
            for interval in valid_times:
                ax.plot(interval, [interval_y, interval_y])
                ax.scatter(
                    interval, [interval_y, interval_y], alpha=0.8, zorder=2
                )
                ax.text(
                    interval[0] + np.diff(interval)[0] / 2,
                    interval_y,
                    epoch,
                    ha="center",
                    va="bottom",
                )

        pos_valid_times = (
            interval_list.set_index("interval_list_name")
            .filter(regex=r"^pos \d+ valid times$", axis=0)
            .valid_times
        ).sort_index(key=lambda index: [int(name.split()[1]) for name in index])
        interval_y = 0
        for epoch, valid_times in zip(pos_valid_times.index, pos_valid_times):
            for interval in valid_times:
                ax.plot(interval, [interval_y, interval_y])
                ax.scatter(
                    interval, [interval_y, interval_y], alpha=0.8, zorder=2
                )
                ax.text(
                    interval[0] + np.diff(interval)[0] / 2,
                    interval_y,
                    epoch.replace(" valid times", ""),
                    ha="center",
                    va="bottom",
                )

        ax.set_ylim((-0.25, 2.25))
        ax.set_yticks(np.arange(3))
        ax.set_yticklabels(["pos valid times", "raw data valid times", "epoch"])
        ax.set_xlabel("Time [s]")
        ax.grid(True)
        if return_fig:
            return fig

    def cautious_insert(self, inserts, **kwargs):
        """On existing primary key, check secondary key and update if needed.

        `replace=True` will attempt to delete/replace the existing entry. When
        the row has a foreign key constraint, this will fail. This method will
        check if an update is needed.

        Parameters
        ----------
        inserts : list of dict
            List of dictionaries to insert.
        **kwargs : dict
            Additional keyword arguments to pass to `insert`.
        """
        if not isinstance(inserts, list):
            inserts = [inserts]
        if not isinstance(inserts[0], dict):
            raise ValueError("Input must be a list of dictionaries.")

        pk = self.heading.primary_key

        def pk_match(row):
            match = self & {k: v for k, v in row.items() if k in pk}
            return match.fetch(as_dict=True)[0] if match else None

        def sk_match(new, old):
            return (
                np.array_equal(new["valid_times"], old["valid_times"])
                and new["pipeline"] == old["pipeline"]
            )

        basic_inserts, need_update = [], []
        for row in inserts:
            existing = pk_match(row)
            if not existing:  # if no existing entry, insert
                basic_inserts.append(row)
            elif existing and not sk_match(row, existing):  # diff sk, update
                need_update.append(row)

        self.insert(basic_inserts, **kwargs)
        for row in need_update:
            self.update1(row)

    def cleanup(self, dry_run=True):
        """Clean up orphaned IntervalList entries."""
        orphans = self - get_child_tables(self)
        if dry_run:
            return orphans
        orphans.super_delete(warn=False)

insert_from_nwbfile(nwbf, *, nwb_file_name) classmethod

Add each entry in the NWB file epochs table to the IntervalList.

The interval list name for each epoch is set to the first tag for the epoch. If the epoch has no tags, then 'interval_x' will be used as the interval list name, where x is the index (0-indexed) of the epoch in the epochs table. The start time and stop time of the epoch are stored in the valid_times field as a numpy array of [start time, stop time] for each epoch.

Parameters:

Name Type Description Default
nwbf NWBFile

The source NWB file object.

 required
nwb_file_name str

The file name of the NWB file, used as a primary key to the Session table.

 required
Source code in src/spyglass/common/common_interval.py 
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
@classmethod
def insert_from_nwbfile(cls, nwbf: NWBFile, *, nwb_file_name: str):
    """Add each entry in the NWB file epochs table to the IntervalList.

    The interval list name for each epoch is set to the first tag for the
    epoch. If the epoch has no tags, then 'interval_x' will be used as the
    interval list name, where x is the index (0-indexed) of the epoch in the
    epochs table. The start time and stop time of the epoch are stored in
    the valid_times field as a numpy array of [start time, stop time] for
    each epoch.

    Parameters
    ----------
    nwbf : pynwb.NWBFile
        The source NWB file object.
    nwb_file_name : str
        The file name of the NWB file, used as a primary key to the Session
        table.
    """
    if nwbf.epochs is None:
        logger.info("No epochs found in NWB file.")
        return

    epochs = nwbf.epochs.to_dataframe()

    # Create a list of dictionaries to insert
    epoch_inserts = epochs.apply(
        lambda epoch_data: {
            "nwb_file_name": nwb_file_name,
            "interval_list_name": (
                epoch_data.tags[0]
                if epoch_data.tags
                else f"interval_{epoch_data.name}"
            ),
            "valid_times": np.asarray(
                [[epoch_data.start_time, epoch_data.stop_time]]
            ),
        },
        axis=1,
    ).tolist()

    cls.insert(epoch_inserts, skip_duplicates=True)

plot_intervals(figsize=(20, 5), return_fig=False)

Plot the intervals in the interval list.

Source code in src/spyglass/common/common_interval.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def plot_intervals(self, figsize=(20, 5), return_fig=False):
    """Plot the intervals in the interval list."""
    interval_list = pd.DataFrame(self)
    fig, ax = plt.subplots(figsize=figsize)
    interval_count = 0
    for row in interval_list.itertuples(index=False):
        for interval in row.valid_times:
            ax.plot(interval, [interval_count, interval_count])
            ax.scatter(
                interval,
                [interval_count, interval_count],
                alpha=0.8,
                zorder=2,
            )
        interval_count += 1
    ax.set_yticks(np.arange(interval_list.shape[0]))
    ax.set_yticklabels(interval_list.interval_list_name)
    ax.set_xlabel("Time [s]")
    ax.grid(True)
    if return_fig:
        return fig

plot_epoch_pos_raw_intervals(figsize=(20, 5), return_fig=False)

Plot an epoch's position, raw data, and valid times intervals.

Source code in src/spyglass/common/common_interval.py 
 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
def plot_epoch_pos_raw_intervals(self, figsize=(20, 5), return_fig=False):
    """Plot an epoch's position, raw data, and valid times intervals."""
    interval_list = pd.DataFrame(self)
    fig, ax = plt.subplots(figsize=(30, 3))

    raw_data_valid_times = interval_list.loc[
        interval_list.interval_list_name == "raw data valid times"
    ].valid_times
    interval_y = 1

    for interval in np.asarray(raw_data_valid_times)[0]:
        ax.plot(interval, [interval_y, interval_y])
        ax.scatter(interval, [interval_y, interval_y], alpha=0.8, zorder=2)

    epoch_valid_times = (
        interval_list.set_index("interval_list_name")
        .filter(regex=r"^[0-9]", axis=0)
        .valid_times
    )
    interval_y = 2
    for epoch, valid_times in zip(
        epoch_valid_times.index, epoch_valid_times
    ):
        for interval in valid_times:
            ax.plot(interval, [interval_y, interval_y])
            ax.scatter(
                interval, [interval_y, interval_y], alpha=0.8, zorder=2
            )
            ax.text(
                interval[0] + np.diff(interval)[0] / 2,
                interval_y,
                epoch,
                ha="center",
                va="bottom",
            )

    pos_valid_times = (
        interval_list.set_index("interval_list_name")
        .filter(regex=r"^pos \d+ valid times$", axis=0)
        .valid_times
    ).sort_index(key=lambda index: [int(name.split()[1]) for name in index])
    interval_y = 0
    for epoch, valid_times in zip(pos_valid_times.index, pos_valid_times):
        for interval in valid_times:
            ax.plot(interval, [interval_y, interval_y])
            ax.scatter(
                interval, [interval_y, interval_y], alpha=0.8, zorder=2
            )
            ax.text(
                interval[0] + np.diff(interval)[0] / 2,
                interval_y,
                epoch.replace(" valid times", ""),
                ha="center",
                va="bottom",
            )

    ax.set_ylim((-0.25, 2.25))
    ax.set_yticks(np.arange(3))
    ax.set_yticklabels(["pos valid times", "raw data valid times", "epoch"])
    ax.set_xlabel("Time [s]")
    ax.grid(True)
    if return_fig:
        return fig

cautious_insert(inserts, **kwargs)

On existing primary key, check secondary key and update if needed.

replace=True will attempt to delete/replace the existing entry. When the row has a foreign key constraint, this will fail. This method will check if an update is needed.

Parameters:

Name Type Description Default
inserts list of dict

List of dictionaries to insert.

 required
**kwargs dict

Additional keyword arguments to pass to insert.

 {}
Source code in src/spyglass/common/common_interval.py 
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
def cautious_insert(self, inserts, **kwargs):
    """On existing primary key, check secondary key and update if needed.

    `replace=True` will attempt to delete/replace the existing entry. When
    the row has a foreign key constraint, this will fail. This method will
    check if an update is needed.

    Parameters
    ----------
    inserts : list of dict
        List of dictionaries to insert.
    **kwargs : dict
        Additional keyword arguments to pass to `insert`.
    """
    if not isinstance(inserts, list):
        inserts = [inserts]
    if not isinstance(inserts[0], dict):
        raise ValueError("Input must be a list of dictionaries.")

    pk = self.heading.primary_key

    def pk_match(row):
        match = self & {k: v for k, v in row.items() if k in pk}
        return match.fetch(as_dict=True)[0] if match else None

    def sk_match(new, old):
        return (
            np.array_equal(new["valid_times"], old["valid_times"])
            and new["pipeline"] == old["pipeline"]
        )

    basic_inserts, need_update = [], []
    for row in inserts:
        existing = pk_match(row)
        if not existing:  # if no existing entry, insert
            basic_inserts.append(row)
        elif existing and not sk_match(row, existing):  # diff sk, update
            need_update.append(row)

    self.insert(basic_inserts, **kwargs)
    for row in need_update:
        self.update1(row)

cleanup(dry_run=True)

Clean up orphaned IntervalList entries.

Source code in src/spyglass/common/common_interval.py 
204
205
206
207
208
209
def cleanup(self, dry_run=True):
    """Clean up orphaned IntervalList entries."""
    orphans = self - get_child_tables(self)
    if dry_run:
        return orphans
    orphans.super_delete(warn=False)

intervals_by_length(interval_list, min_length=0.0, max_length=10000000000.0)

Select intervals of certain lengths from an interval list.

Parameters:

Name Type Description Default
interval_list array_like

Each element is (start time, stop time), i.e. an interval in seconds.

 required
min_length float

Minimum interval length in seconds. Defaults to 0.0.

 0.0
max_length float

Maximum interval length in seconds. Defaults to 1e10.

 10000000000.0
Source code in src/spyglass/common/common_interval.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def intervals_by_length(interval_list, min_length=0.0, max_length=1e10):
    """Select intervals of certain lengths from an interval list.

    Parameters
    ----------
    interval_list : array_like
        Each element is (start time, stop time), i.e. an interval in seconds.
    min_length : float, optional
        Minimum interval length in seconds. Defaults to 0.0.
    max_length : float, optional
        Maximum interval length in seconds. Defaults to 1e10.
    """
    lengths = np.ravel(np.diff(interval_list))
    return interval_list[
        np.logical_and(lengths > min_length, lengths < max_length)
    ]

interval_list_contains_ind(interval_list, timestamps)

Find indices of list of timestamps contained in an interval list.

Parameters:

Name Type Description Default
interval_list array_like

Each element is (start time, stop time), i.e. an interval in seconds.

 required
timestamps array_like
required
Source code in src/spyglass/common/common_interval.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def interval_list_contains_ind(interval_list, timestamps):
    """Find indices of list of timestamps contained in an interval list.

    Parameters
    ----------
    interval_list : array_like
        Each element is (start time, stop time), i.e. an interval in seconds.
    timestamps : array_like
    """
    ind = []
    for interval in interval_list:
        ind += np.ravel(
            np.argwhere(
                np.logical_and(
                    timestamps >= interval[0], timestamps <= interval[1]
                )
            )
        ).tolist()
    return np.asarray(ind)

interval_list_contains(interval_list, timestamps)

Find timestamps that are contained in an interval list.

Parameters:

Name Type Description Default
interval_list array_like

Each element is (start time, stop time), i.e. an interval in seconds.

 required
timestamps array_like
required
Source code in src/spyglass/common/common_interval.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def interval_list_contains(interval_list, timestamps):
    """Find timestamps that are contained in an interval list.

    Parameters
    ----------
    interval_list : array_like
        Each element is (start time, stop time), i.e. an interval in seconds.
    timestamps : array_like
    """
    ind = []
    for interval in interval_list:
        ind += np.ravel(
            np.argwhere(
                np.logical_and(
                    timestamps >= interval[0], timestamps <= interval[1]
                )
            )
        ).tolist()
    return timestamps[ind]

interval_list_excludes_ind(interval_list, timestamps)

Find indices of timestamps that are not contained in an interval list.

Parameters:

Name Type Description Default
interval_list array_like

Each element is (start time, stop time), i.e. an interval in seconds.

 required
timestamps array_like
required
Source code in src/spyglass/common/common_interval.py 
272
273
274
275
276
277
278
279
280
281
282
283
def interval_list_excludes_ind(interval_list, timestamps):
    """Find indices of timestamps that are not contained in an interval list.

    Parameters
    ----------
    interval_list : array_like
        Each element is (start time, stop time), i.e. an interval in seconds.
    timestamps : array_like
    """

    contained_inds = interval_list_contains_ind(interval_list, timestamps)
    return np.setdiff1d(np.arange(len(timestamps)), contained_inds)

interval_list_excludes(interval_list, timestamps)

Find timestamps that are not contained in an interval list.

Parameters:

Name Type Description Default
interval_list array_like

Each element is (start time, stop time), i.e. an interval in seconds.

 required
timestamps array_like
required
Source code in src/spyglass/common/common_interval.py 
286
287
288
289
290
291
292
293
294
295
296
def interval_list_excludes(interval_list, timestamps):
    """Find timestamps that are not contained in an interval list.

    Parameters
    ----------
    interval_list : array_like
        Each element is (start time, stop time), i.e. an interval in seconds.
    timestamps : array_like
    """
    contained_times = interval_list_contains(interval_list, timestamps)
    return np.setdiff1d(timestamps, contained_times)

consolidate_intervals(interval_list)

Consolidate overlapping intervals in an interval list.

Source code in src/spyglass/common/common_interval.py 
299
300
301
302
303
304
305
306
307
308
309
310
def consolidate_intervals(interval_list):
    """Consolidate overlapping intervals in an interval list."""
    if interval_list.ndim == 1:
        interval_list = np.expand_dims(interval_list, 0)
    else:
        interval_list = interval_list[np.argsort(interval_list[:, 0])]
        interval_list = reduce(_union_concat, interval_list)
        # the following check is needed in the case where the interval list is a
        # single element (behavior of reduce)
        if interval_list.ndim == 1:
            interval_list = np.expand_dims(interval_list, 0)
    return interval_list

interval_list_intersect(interval_list1, interval_list2, min_length=0)

Finds the intersections between two interval lists

Each interval is (start time, stop time)

Parameters:

Name Type Description Default
interval_list1 np.array, (N,2) where N = number of intervals
required
interval_list2 np.array, (N,2) where N = number of intervals
required
min_length float, optional.

Minimum length of intervals to include, default 0

 0

Returns:

Name Type Description
interval_list (array, (N, 2))
Source code in src/spyglass/common/common_interval.py 
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
def interval_list_intersect(interval_list1, interval_list2, min_length=0):
    """Finds the intersections between two interval lists

    Each interval is (start time, stop time)

    Parameters
    ----------
    interval_list1 : np.array, (N,2) where N = number of intervals
    interval_list2 : np.array, (N,2) where N = number of intervals
    min_length : float, optional.
        Minimum length of intervals to include, default 0

    Returns
    -------
    interval_list: np.array, (N,2)
    """

    # Consolidate interval lists to disjoint int'ls by sorting & applying union
    interval_list1 = consolidate_intervals(interval_list1)
    interval_list2 = consolidate_intervals(interval_list2)

    # then do pairwise comparison and collect intersections
    intersecting_intervals = [
        _intersection(interval2, interval1)
        for interval2 in interval_list2
        for interval1 in interval_list1
        if _intersection(interval2, interval1) is not None
    ]

    # if no intersection, then return an empty list
    if not intersecting_intervals:
        return []

    intersecting_intervals = np.asarray(intersecting_intervals)
    intersecting_intervals = intersecting_intervals[
        np.argsort(intersecting_intervals[:, 0])
    ]

    return intervals_by_length(intersecting_intervals, min_length=min_length)

union_adjacent_index(interval1, interval2)

Union index-adjacent intervals. If not adjacent, just concatenate.

e.g. [a,b] and [b+1, c] is converted to [a,c]

Parameters:

Name Type Description Default
interval1 array
required
interval2 array
required
Source code in src/spyglass/common/common_interval.py 
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
def union_adjacent_index(interval1, interval2):
    """Union index-adjacent intervals. If not adjacent, just concatenate.

    e.g. [a,b] and [b+1, c] is converted to [a,c]

    Parameters
    ----------
    interval1 : np.array
    interval2 : np.array
    """
    interval1 = np.atleast_2d(interval1)
    interval2 = np.atleast_2d(interval2)

    if (
        interval1[-1][1] + 1 == interval2[0][0]
        or interval2[0][1] + 1 == interval1[-1][0]
    ):
        x = np.array(
            [
                [
                    np.min([interval1[-1][0], interval2[0][0]]),
                    np.max([interval1[-1][1], interval2[0][1]]),
                ]
            ]
        )
        return np.concatenate((interval1[:-1], x), axis=0)
    else:
        return np.concatenate((interval1, interval2), axis=0)

interval_list_union(interval_list1, interval_list2, min_length=0.0, max_length=10000000000.0)

Finds the union (all times in one or both) for two interval lists

Parameters:

Name Type Description Default
interval_list1 ndarray

The first interval list [start, stop]

 required
interval_list2 ndarray

The second interval list [start, stop]

 required
min_length float

Minimum length of interval for inclusion in output, default 0.0

 0.0
max_length float

Maximum length of interval for inclusion in output, default 1e10

 10000000000.0

Returns:

Type Description
ndarray

Array of intervals [start, stop]
Source code in src/spyglass/common/common_interval.py 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
def interval_list_union(
    interval_list1: np.ndarray,
    interval_list2: np.ndarray,
    min_length: float = 0.0,
    max_length: float = 1e10,
) -> np.ndarray:
    """Finds the union (all times in one or both) for two interval lists

    Parameters
    ----------
    interval_list1 : np.ndarray
        The first interval list [start, stop]
    interval_list2 : np.ndarray
        The second interval list [start, stop]
    min_length : float, optional
        Minimum length of interval for inclusion in output, default 0.0
    max_length : float, optional
        Maximum length of interval for inclusion in output, default 1e10

    Returns
    -------
    np.ndarray
        Array of intervals [start, stop]
    """

    il1, il1_start_end = _parallel_union(interval_list1)
    il2, il2_start_end = _parallel_union(interval_list2)

    # Concatenate the two lists so we can resort the intervals and apply the
    # same sorting to the start-end arrays
    combined_intervals = np.concatenate((il1, il2))
    ss = np.concatenate((il1_start_end, il2_start_end))
    sort_ind = np.argsort(combined_intervals)
    combined_intervals = combined_intervals[sort_ind]

    # a cumulative sum of 1 indicates the beginning of a joint interval; a
    # cumulative sum of 0 indicates the end
    union_starts = np.ravel(np.array(np.where(np.cumsum(ss[sort_ind]) == 1)))
    union_stops = np.ravel(np.array(np.where(np.cumsum(ss[sort_ind]) == 0)))
    union = [
        [combined_intervals[start], combined_intervals[stop]]
        for start, stop in zip(union_starts, union_stops)
    ]

    return np.asarray(union)

interval_list_censor(interval_list, timestamps)

Returns new interval list that starts/ends at first/last timestamp

Parameters:

Name Type Description Default
interval_list numpy array of intervals [start, stop]

interval list from IntervalList valid times

 required
timestamps numpy array or list
required

Returns:

Type Description
interval_list (numpy array of intervals [start, stop])
Source code in src/spyglass/common/common_interval.py 
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
def interval_list_censor(interval_list, timestamps):
    """Returns new interval list that starts/ends at first/last timestamp

    Parameters
    ----------
    interval_list : numpy array of intervals [start, stop]
        interval list from IntervalList valid times
    timestamps : numpy array or list

    Returns
    -------
    interval_list (numpy array of intervals [start, stop])
    """
    # check that all timestamps are in the interval list
    if len(interval_list_contains_ind(interval_list, timestamps)) != len(
        timestamps
    ):
        raise ValueError("Interval_list must contain all timestamps")

    timestamps_interval = np.asarray([[timestamps[0], timestamps[-1]]])
    return interval_list_intersect(interval_list, timestamps_interval)

interval_from_inds(list_frames)

Converts a list of indices to a list of intervals.

e.g. [2,3,4,6,7,8,9,10] -> [[2,4],[6,10]]

Parameters:

Name Type Description Default
list_frames array_like of int
required
Source code in src/spyglass/common/common_interval.py 
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
def interval_from_inds(list_frames):
    """Converts a list of indices to a list of intervals.

    e.g. [2,3,4,6,7,8,9,10] -> [[2,4],[6,10]]

    Parameters
    ----------
    list_frames : array_like of int
    """
    list_frames = np.unique(list_frames)
    interval_list = []
    for key, group in itertools.groupby(
        enumerate(list_frames), lambda t: t[1] - t[0]
    ):
        group = list(group)
        interval_list.append([group[0][1], group[-1][1]])
    return np.asarray(interval_list)

interval_set_difference_inds(intervals1, intervals2)

e.g. intervals1 = [(0, 5), (8, 10)] intervals2 = [(1, 2), (3, 4), (6, 9)]

result = [(0, 1), (4, 5), (9, 10)]

Parameters:

Name Type Description Default
intervals1 _type_

description

 required
intervals2 _type_

description

 required

Returns:

Type Description
_type_

description
Source code in src/spyglass/common/common_interval.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
def interval_set_difference_inds(intervals1, intervals2):
    """
    e.g.
    intervals1 = [(0, 5), (8, 10)]
    intervals2 = [(1, 2), (3, 4), (6, 9)]

    result = [(0, 1), (4, 5), (9, 10)]

    Parameters
    ----------
    intervals1 : _type_
        _description_
    intervals2 : _type_
        _description_

    Returns
    -------
    _type_
        _description_
    """
    result = []
    i = j = 0
    while i < len(intervals1) and j < len(intervals2):
        if intervals1[i][1] <= intervals2[j][0]:
            result.append(intervals1[i])
            i += 1
        elif intervals2[j][1] <= intervals1[i][0]:
            j += 1
        else:
            if intervals1[i][0] < intervals2[j][0]:
                result.append((intervals1[i][0], intervals2[j][0]))
            if intervals1[i][1] > intervals2[j][1]:
                intervals1[i] = (intervals2[j][1], intervals1[i][1])
                j += 1
            else:
                i += 1
    result += intervals1[i:]
    return result

interval_list_complement(intervals1, intervals2, min_length=0.0)

Finds intervals in intervals1 that are not in intervals2

Parameters:

Name Type Description Default
min_length float

Minimum interval length in seconds. Defaults to 0.0.

 0.0
Source code in src/spyglass/common/common_interval.py 
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
def interval_list_complement(intervals1, intervals2, min_length=0.0):
    """
    Finds intervals in intervals1 that are not in intervals2

    Parameters
    ----------
    min_length : float, optional
        Minimum interval length in seconds. Defaults to 0.0.
    """

    result = []

    for start1, end1 in intervals1:
        subtracted = [(start1, end1)]

        for start2, end2 in intervals2:
            new_subtracted = []

            for s, e in subtracted:
                if start2 <= s and e <= end2:
                    continue

                if e <= start2 or end2 <= s:
                    new_subtracted.append((s, e))
                    continue

                if start2 > s:
                    new_subtracted.append((s, start2))

                if end2 < e:
                    new_subtracted.append((end2, e))

            subtracted = new_subtracted

        result.extend(subtracted)

    return intervals_by_length(
        np.asarray(result), min_length=min_length, max_length=1e100
    )