Skip to content

deeplabcut.refine_training_dataset

Modules:

Name Description
auxfun_multianimal

DeepLabCut2.0 Toolbox (deeplabcut.org)
auxiliaryfunctions

DeepLabCut2.0 Toolbox (deeplabcut.org)
conversioncode
frameselectiontools

DeepLabCut2.0 Toolbox (deeplabcut.org)
inferenceutils
outlier_frames
stitch
tracklets
visualization

DeepLabCut2.0 Toolbox (deeplabcut.org)

Classes:

Name Description
TrackletManager
VideoWriter

Functions:

Name Description
PlottingSingleFrame

Label frame and save under imagename / this is already cropped (for clip)
PlottingSingleFramecv2

Label frame and save under imagename / cap is not already cropped.
attempt_to_add_video

Add new videos to the config file at any stage of the project.
collect_video_paths

Collects video paths from a given set of data paths: directories, files, or a mix
columnwise_spline_interp

Perform cubic spline interpolation over the columns of data. All gaps of size
compute_deviations

Fits Seasonal AutoRegressive Integrated Moving Average with eXogenous regressors
convertparms2start

Creating a start value for sarimax in case of an value error
extract_outlier_frames

Extracts the outlier frames.
find_outliers_in_raw_data

Extract outlier frames from either raw detections or assemblies of multiple
find_outliers_in_raw_detections

Find outlier frames from the raw detections of multiple animals.
merge_datasets

Merge the original training dataset with the newly refined data.
renamed_parameter

Support a renamed keyword argument while warning callers to update.

TrackletManager

Methods:

Name Description
__init__

Parameters
Source code in deeplabcut/refine_training_dataset/tracklets.py 
 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
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
302
303
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
class TrackletManager:
    def __init__(self, config, min_swap_len=2, min_tracklet_len=2, max_gap=0):
        """

        Parameters
        ----------
        config : str
            Path to a configuration file.
        min_swap_len : float, optional (default=2)
            Minimum swap length.
            Swaps shorter than 2 frames are discarded by default.
        min_tracklet_len : float, optional (default=2)
            Minimum tracklet length.
            Tracklets shorter than 2 frames are discarded by default.
        max_gap : int, optional (default = 0).
            Number of frames to consider when filling in missing data.

        Examples
        --------

        manager = TrackletManager(config_path, min_swap_frac=0, min_tracklet_frac=0)

        manager.load_tracklets_from_pickle(filename)
        # Alternatively
        manager.load_tracklets_from_h5(filename)

        manager.find_swapping_bodypart_pairs()
        """
        self.config = config
        self.cfg = auxiliaryfunctions.read_config(config)
        self.min_swap_len = min_swap_len
        self.min_tracklet_len = min_tracklet_len
        self.max_gap = max_gap

        self.filename = ""
        self.data = None
        self.xy = None
        self._xy = None
        self.prob = None
        self.nframes = 0
        self.times = []
        self.scorer = None
        self.bodyparts = []
        self.nindividuals = len(self.cfg["individuals"])
        self.individuals = []
        self.tracklet2id = []
        self.tracklet2bp = []
        self.swapping_pairs = []
        self.swapping_bodyparts = []
        self._label_pairs = None

    def _load_tracklets(self, tracklets, auto_fill):
        header = tracklets.pop("header")
        self.scorer = header.get_level_values("scorer").unique().to_list()
        bodyparts = header.get_level_values("bodyparts")
        bodyparts_multi = [bp for bp in self.cfg["multianimalbodyparts"] if bp in bodyparts]
        bodyparts_single = self.cfg["uniquebodyparts"]
        mask_multi = bodyparts.isin(bodyparts_multi)
        mask_single = bodyparts.isin(bodyparts_single)
        self.bodyparts = list(bodyparts[mask_multi]) * self.nindividuals + list(bodyparts[mask_single])

        # Sort tracklets by length to prioritize greater continuity
        temp = sorted(tracklets.values(), key=len)
        if not len(temp):
            raise OSError("Tracklets are empty.")

        def get_frame_ind(s):
            return int(re.findall(r"\d+", s)[0])

        # Drop tracklets that are too short
        tracklets_sorted = []
        last_frames = []
        for tracklet in temp:
            last_frames.append(get_frame_ind(list(tracklet)[-1]))
            if len(tracklet) > self.min_tracklet_len:
                tracklets_sorted.append(tracklet)
        self.nframes = max(last_frames) + 1
        self.times = np.arange(self.nframes)

        if auto_fill:  # Recursively fill the data containers
            tracklets_multi = np.full(
                (self.nindividuals, self.nframes, len(bodyparts_multi) * 3),
                np.nan,
                np.float16,
            )
            tracklets_single = np.full((self.nframes, len(bodyparts_single) * 3), np.nan, np.float16)
            for _ in trange(len(tracklets_sorted)):
                tracklet = tracklets_sorted.pop()
                inds, temp = zip(*[(get_frame_ind(k), v) for k, v in tracklet.items()], strict=False)
                inds = np.asarray(inds)
                data = np.asarray(temp, dtype=np.float16)
                data_single = data[:, mask_single]
                is_multi = np.isnan(data_single).all()
                if not is_multi:
                    # Where slots are available, copy the data over
                    is_free = np.isnan(tracklets_single[inds])
                    has_data = ~np.isnan(data_single)
                    mask = has_data & is_free
                    rows, cols = np.nonzero(mask)
                    tracklets_single[inds[rows], cols] = data_single[mask]
                    # If about to overwrite data, keep tracklets with highest confidence
                    overwrite = has_data & ~is_free
                    if overwrite.any():
                        rows, cols = np.nonzero(overwrite)
                        more_confident = (data_single[overwrite] > tracklets_single[inds[rows], cols])[2::3]
                        idx = np.flatnonzero(more_confident)
                        for i in idx:
                            sl = slice(i * 3, i * 3 + 3)
                            tracklets_single[inds[rows[sl]], cols[sl]] = data_single[rows[sl], cols[sl]]
                else:
                    is_free = np.isnan(tracklets_multi[:, inds])
                    data_multi = data[:, mask_multi]
                    has_data = ~np.isnan(data_multi)
                    overwrite = has_data & ~is_free
                    overwrite_risk = np.any(overwrite, axis=(1, 2))
                    if overwrite_risk.all():
                        # Squeeze some data into empty slots
                        n_empty = is_free.all(axis=2).sum(axis=1)
                        for ind in np.argsort(n_empty)[::-1]:
                            mask = has_data & is_free
                            current_mask = mask[ind]
                            rows, cols = np.nonzero(current_mask)
                            if rows.size:
                                tracklets_multi[ind, inds[rows], cols] = data_multi[current_mask]
                                is_free[ind, current_mask] = False
                                has_data[current_mask] = False
                        if has_data.any():
                            # For the remaining data, overwrite where we are least confident
                            remaining = data_multi[has_data].reshape((-1, 3))
                            mask3d = np.broadcast_to(has_data, (self.nindividuals,) + has_data.shape)
                            dims, rows, cols = np.nonzero(mask3d)
                            temp = tracklets_multi[dims, inds[rows], cols].reshape((self.nindividuals, -1, 3))
                            diff = remaining - temp
                            # Find keypoints closest to the remaining data
                            # Use Manhattan distance to avoid overflow
                            dist = np.abs(diff[:, :, 0]) + np.abs(diff[:, :, 1])
                            closest = np.argmin(dist, axis=0)
                            # Only overwrite if improving confidence
                            prob = diff[closest, range(len(closest)), 2]
                            better = np.flatnonzero(prob > 0)
                            idx = closest[better]
                            rows, cols = np.nonzero(has_data)
                            for i, j in zip(idx, better, strict=False):
                                sl = slice(j * 3, j * 3 + 3)
                                tracklets_multi[i, inds[rows[sl]], cols[sl]] = remaining.flat[sl]
                    else:
                        rows, cols = np.nonzero(has_data)
                        n = np.argmin(overwrite_risk)
                        tracklets_multi[n, inds[rows], cols] = data_multi[has_data]

            multi = tracklets_multi.swapaxes(0, 1).reshape((self.nframes, -1))
            data = np.c_[multi, tracklets_single].reshape((self.nframes, -1, 3))
            xy = data[:, :, :2].reshape((self.nframes, -1))
            prob = data[:, :, 2].reshape((self.nframes, -1))

            # Fill existing gaps
            missing = np.isnan(xy)
            xy_filled = columnwise_spline_interp(xy, self.max_gap)
            filled = ~np.isnan(xy_filled)
            xy[filled] = xy_filled[filled]
            inds = np.argwhere(missing & filled)
            if inds.size:
                # Retrieve original individual label indices
                inds[:, 1] //= 2
                inds = np.unique(inds, axis=0)
                prob[inds[:, 0], inds[:, 1]] = 0.01
            data[:, :, :2] = xy.reshape((self.nframes, -1, 2))
            data[:, :, 2] = prob
            self.data = data.swapaxes(0, 1)
            self.xy = self.data[:, :, :2]
            self.prob = self.data[:, :, 2]

            # Map a tracklet # to the animal ID it belongs to or the bodypart # it corresponds to.
            self.individuals = self.cfg["individuals"] + (["single"] if len(self.cfg["uniquebodyparts"]) else [])
            self.tracklet2id = [i for i in range(0, self.nindividuals) for _ in bodyparts_multi] + [
                self.nindividuals
            ] * len(bodyparts_single)
            bps = bodyparts_multi + bodyparts_single
            map_ = dict(zip(bps, range(len(bps)), strict=False))
            self.tracklet2bp = [map_[bp] for bp in self.bodyparts[::3]]
            self._label_pairs = self.get_label_pairs()
        else:
            tracklets_raw = np.full(
                (len(tracklets_sorted), self.nframes, len(bodyparts)),
                np.nan,
                np.float16,
            )
            for n, tracklet in enumerate(tracklets_sorted[::-1]):
                for frame, data in tracklet.items():
                    i = get_frame_ind(frame)
                    tracklets_raw[n, i] = data
            self.data = tracklets_raw.swapaxes(0, 1).reshape((self.nframes, -1, 3)).swapaxes(0, 1)
            self.xy = self.data[:, :, :2]
            self.prob = self.data[:, :, 2]
            self.tracklet2id = self.tracklet2bp = [0] * self.data.shape[0]

    def load_tracklets_from_pickle(self, filename, auto_fill=True):
        self.filename = filename
        with open(filename, "rb") as file:
            tracklets = pickle.load(file)
        self._load_tracklets(tracklets, auto_fill)
        self._xy = self.xy.copy()

    def load_tracklets_from_hdf(self, filename):
        self.filename = filename
        df = pd.read_hdf(filename)

        # Fill existing gaps
        data = df.to_numpy()
        mask = ~df.columns.get_level_values(level="coords").str.contains("likelihood")
        xy = data[:, mask]
        prob = data[:, ~mask]
        missing = np.isnan(xy)
        xy_filled = columnwise_spline_interp(xy, self.max_gap)
        filled = ~np.isnan(xy_filled)
        xy[filled] = xy_filled[filled]
        inds = np.argwhere(missing & filled)
        if inds.size:
            # Retrieve original individual label indices
            inds[:, 1] //= 2
            inds = np.unique(inds, axis=0)
            prob[inds[:, 0], inds[:, 1]] = 0.01
        data[:, mask] = xy
        data[:, ~mask] = prob
        df = pd.DataFrame(data, index=df.index, columns=df.columns)

        idx = df.columns
        self.scorer = idx.get_level_values("scorer").unique().to_list()
        self.bodyparts = idx.get_level_values("bodyparts")
        self.nframes = len(df)
        self.times = np.arange(self.nframes)
        self.data = df.values.reshape((self.nframes, -1, 3)).swapaxes(0, 1)
        self.xy = self.data[:, :, :2]
        self.prob = self.data[:, :, 2]
        individuals = idx.get_level_values("individuals")
        self.individuals = individuals.unique().to_list()
        self.tracklet2id = individuals.map(
            dict(zip(self.individuals, range(len(self.individuals)), strict=False))
        ).tolist()[::3]
        bodyparts = self.bodyparts.unique()
        self.tracklet2bp = self.bodyparts.map(dict(zip(bodyparts, range(len(bodyparts)), strict=False))).tolist()[::3]
        self._label_pairs = list(idx.droplevel(["scorer", "coords"]).unique())
        self._xy = self.xy.copy()

    def calc_completeness(self, xy, by_individual=False):
        comp = np.sum(~np.isnan(xy).any(axis=2), axis=1)
        if by_individual:
            inds = np.insert(np.diff(self.tracklet2id), 0, 1)
            comp = np.add.reduceat(comp, np.flatnonzero(inds))
        return comp

    def to_num_bodypart(self, ind):
        return self.tracklet2bp[ind]

    def to_num_individual(self, ind):
        return self.tracklet2id[ind]

    def get_non_nan_elements(self, at):
        data = self.xy[:, at]
        mask = ~np.isnan(data).any(axis=1)
        return data[mask], mask, np.flatnonzero(mask)

    def swap_tracklets(self, track1, track2, inds):
        self.xy[np.ix_([track1, track2], inds)] = self.xy[np.ix_([track2, track1], inds)]
        self.prob[np.ix_([track1, track2], inds)] = self.prob[np.ix_([track2, track1], inds)]
        self.tracklet2bp[track1], self.tracklet2bp[track2] = (
            self.tracklet2bp[track2],
            self.tracklet2bp[track1],
        )

    def find_swapping_bodypart_pairs(self, force_find=False):
        if not self.swapping_pairs or force_find:
            sub = self.xy[:, np.newaxis] - self.xy  # Broadcasting for efficient subtraction of X and Y coordinates
            with np.errstate(invalid="ignore"):  # Get rid of annoying warnings when comparing with NaNs
                pos = sub > 0
                neg = sub <= 0
                down = neg[:, :, 1:] & pos[:, :, :-1]
                up = pos[:, :, 1:] & neg[:, :, :-1]
                zero_crossings = down | up
            # ID swaps occur when X and Y simultaneously intersect each other.
            self.tracklet_swaps = zero_crossings.all(axis=3)
            cross = self.tracklet_swaps.sum(axis=2) > self.min_swap_len
            mat = np.tril(cross)
            temp_pairs = np.where(mat)
            # Get only those bodypart pairs that belong to different individuals
            pairs = []
            for a, b in zip(*temp_pairs, strict=False):
                if self.tracklet2id[a] != self.tracklet2id[b]:
                    pairs.append((a, b))
            self.swapping_pairs = pairs
            self.swapping_bodyparts = np.unique(pairs).tolist()

    def get_swap_indices(self, tracklet1, tracklet2):
        return np.flatnonzero(self.tracklet_swaps[tracklet1, tracklet2])

    def get_nonoverlapping_segments(self, tracklet1, tracklet2):
        swap_inds = self.get_swap_indices(tracklet1, tracklet2)
        inds = np.insert(swap_inds, [0, len(swap_inds)], [0, self.nframes])
        mask = np.ones_like(self.times, dtype=bool)
        for i, j in zip(inds[::2], inds[1::2], strict=False):
            mask[i:j] = False
        return mask

    def flatten_data(self):
        data = np.concatenate((self.xy, np.expand_dims(self.prob, axis=2)), axis=2)
        return data.swapaxes(0, 1).reshape((self.nframes, -1))

    def format_multiindex(self):
        scorer = self.scorer * len(self.bodyparts)
        map_ = dict(zip(range(len(self.individuals)), self.individuals, strict=False))
        individuals = [map_[ind] for ind in self.tracklet2id for _ in range(3)]
        coords = ["x", "y", "likelihood"] * len(self.tracklet2id)
        return pd.MultiIndex.from_arrays(
            [scorer, individuals, self.bodyparts, coords],
            names=["scorer", "individuals", "bodyparts", "coords"],
        )

    def get_label_pairs(self):
        return list(self.format_multiindex().droplevel(["scorer", "coords"]).unique())

    def format_data(self):
        columns = self.format_multiindex()
        return pd.DataFrame(self.flatten_data(), columns=columns, index=self.times)

    def find_edited_frames(self):
        mask = np.isclose(self.xy, self._xy, equal_nan=True).all(axis=(0, 2))
        return np.flatnonzero(~mask)

    def save(self, output_name="", *args):
        df = self.format_data()
        if not output_name:
            output_name = self.filename.replace("pickle", "h5")
        df.to_hdf(output_name, key="df_with_missing", format="table", mode="w")

__init__ 

__init__(config, min_swap_len=2, min_tracklet_len=2, max_gap=0)
Parameters

config : str Path to a configuration file. min_swap_len : float, optional (default=2) Minimum swap length. Swaps shorter than 2 frames are discarded by default. min_tracklet_len : float, optional (default=2) Minimum tracklet length. Tracklets shorter than 2 frames are discarded by default. max_gap : int, optional (default = 0). Number of frames to consider when filling in missing data.

Examples

manager = TrackletManager(config_path, min_swap_frac=0, min_tracklet_frac=0)

manager.load_tracklets_from_pickle(filename)

Alternatively

manager.load_tracklets_from_h5(filename)

manager.find_swapping_bodypart_pairs()

Source code in deeplabcut/refine_training_dataset/tracklets.py 
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
def __init__(self, config, min_swap_len=2, min_tracklet_len=2, max_gap=0):
    """

    Parameters
    ----------
    config : str
        Path to a configuration file.
    min_swap_len : float, optional (default=2)
        Minimum swap length.
        Swaps shorter than 2 frames are discarded by default.
    min_tracklet_len : float, optional (default=2)
        Minimum tracklet length.
        Tracklets shorter than 2 frames are discarded by default.
    max_gap : int, optional (default = 0).
        Number of frames to consider when filling in missing data.

    Examples
    --------

    manager = TrackletManager(config_path, min_swap_frac=0, min_tracklet_frac=0)

    manager.load_tracklets_from_pickle(filename)
    # Alternatively
    manager.load_tracklets_from_h5(filename)

    manager.find_swapping_bodypart_pairs()
    """
    self.config = config
    self.cfg = auxiliaryfunctions.read_config(config)
    self.min_swap_len = min_swap_len
    self.min_tracklet_len = min_tracklet_len
    self.max_gap = max_gap

    self.filename = ""
    self.data = None
    self.xy = None
    self._xy = None
    self.prob = None
    self.nframes = 0
    self.times = []
    self.scorer = None
    self.bodyparts = []
    self.nindividuals = len(self.cfg["individuals"])
    self.individuals = []
    self.tracklet2id = []
    self.tracklet2bp = []
    self.swapping_pairs = []
    self.swapping_bodyparts = []
    self._label_pairs = None

VideoWriter

Bases: VideoReader

Methods:

Name Description
shorten

Shorten the video from start to end.
split

Split a video into several shorter ones of equal duration.
Source code in deeplabcut/utils/auxfun_videos.py 
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
302
303
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
class VideoWriter(VideoReader):
    def __init__(self, video_path, codec="h264", dpi=100, fps=None):
        super().__init__(video_path)
        self.codec = codec
        self.dpi = dpi
        if fps:
            self.fps = fps

    def shorten(self, start, end, suffix="short", dest_folder=None, validate_inputs=True):
        """Shorten the video from start to end.

        Parameter
        ----------
        start: str
            Time formatted in hours:minutes:seconds, where shortened video shall start.

        end: str
            Time formatted in hours:minutes:seconds, where shortened video shall end.

        suffix: str, optional
            String added to the name of the shortened video ('short' by default).

        dest_folder: str, optional
            Folder the video is saved into (by default, same as the original video)

        Returns
        -------
        str
            Full path to the shortened video
        """

        def validate_timestamp(stamp):
            if not isinstance(stamp, str):
                raise ValueError("Timestamp should be a string formatted as hours:minutes:seconds.")
            time = datetime.datetime.strptime(stamp, "%H:%M:%S").time()
            # The above already raises a ValueError if formatting is wrong
            seconds = (time.hour * 60 + time.minute) * 60 + time.second
            if seconds > self.calc_duration():
                raise ValueError("Timestamps must not exceed the video duration.")

        if validate_inputs:
            for stamp in start, end:
                validate_timestamp(stamp)

        output_path = self.make_output_path(suffix, dest_folder)
        command = f'ffmpeg -n -i "{self.video_path}" -ss {start} -to {end} -c:a copy "{output_path}"'
        subprocess.call(command, shell=True)
        return output_path

    def split(self, n_splits, suffix="split", dest_folder=None):
        """Split a video into several shorter ones of equal duration.

        Parameters
        ----------
        n_splits : int
            Number of shorter videos to produce

        suffix: str, optional
            String added to the name of the splits ('short' by default).

        dest_folder: str, optional
            Folder the video splits are saved into (by default, same as the original video)

        Returns
        -------
        list
            Paths of the video splits
        """
        if not n_splits > 1:
            raise ValueError("The video should at least be split in half.")
        chunk_dur = self.calc_duration() / n_splits
        splits = np.arange(n_splits + 1) * chunk_dur

        def time_formatter(val):
            return str(datetime.timedelta(seconds=val))

        clips = []
        for n, (start, end) in enumerate(zip(splits, splits[1:], strict=False), start=1):
            clips.append(
                self.shorten(
                    time_formatter(start),
                    time_formatter(end),
                    f"{suffix}{n}",
                    dest_folder,
                    validate_inputs=False,
                )
            )
        return clips

    def crop(self, suffix="crop", dest_folder=None):
        x1, _, y1, _ = self.get_bbox()
        output_path = self.make_output_path(suffix, dest_folder)
        command = (
            f'ffmpeg -n -i "{self.video_path}" '
            f"-filter:v crop={self.width}:{self.height}:{x1}:{y1} "
            f'-c:a copy "{output_path}"'
        )
        subprocess.call(command, shell=True)
        return output_path

    def rotate(self, angle, rotatecw="Arbitrary", suffix="rotated", dest_folder=None):
        output_path = self.make_output_path(suffix, dest_folder)
        command = f'ffmpeg -n -i "{self.video_path}" -vf '
        if rotatecw == "Arbitrary":
            angle = np.deg2rad(angle)
            command += f"rotate={angle} "
        elif rotatecw == "Yes":
            command += "transpose=1 "
        else:
            raise ValueError("Unknown rotation direction.")

        command += f'-c:a copy "{output_path}"'
        subprocess.call(command, shell=True)
        return output_path

    def rescale(
        self,
        width,
        height=-1,
        rotatecw="No",
        angle=0.0,
        suffix="rescale",
        dest_folder=None,
    ):
        output_path = self.make_output_path(suffix, dest_folder)
        command = f'ffmpeg -n -i "{self.video_path}" -filter:v "scale={width}:{height}{{}}" -c:a copy "{output_path}"'
        # Rotate, see: https://stackoverflow.com/questions/3937387/rotating-videos-with-ffmpeg
        # interesting option to just update metadata.
        if rotatecw == "Arbitrary":
            angle = np.deg2rad(angle)
            command = command.format(f", rotate={angle}")
        elif rotatecw == "Yes":
            command = command.format(", transpose=1")
        else:
            command = command.format("")
        subprocess.call(command, shell=True)
        return output_path

    @staticmethod
    def write_frame(frame, where):
        cv2.imwrite(where, frame[..., ::-1])

    def make_output_path(self, suffix, dest_folder):
        if not dest_folder:
            dest_folder = self.directory
        return os.path.join(dest_folder, f"{self.name}{suffix}{self.format}")

shorten 

shorten(start, end, suffix='short', dest_folder=None, validate_inputs=True)

Shorten the video from start to end.

Parameter

start: str Time formatted in hours:minutes:seconds, where shortened video shall start.

str

Time formatted in hours:minutes:seconds, where shortened video shall end.

str, optional

String added to the name of the shortened video ('short' by default).

str, optional

Folder the video is saved into (by default, same as the original video)

Returns

str Full path to the shortened video

Source code in deeplabcut/utils/auxfun_videos.py 
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
def shorten(self, start, end, suffix="short", dest_folder=None, validate_inputs=True):
    """Shorten the video from start to end.

    Parameter
    ----------
    start: str
        Time formatted in hours:minutes:seconds, where shortened video shall start.

    end: str
        Time formatted in hours:minutes:seconds, where shortened video shall end.

    suffix: str, optional
        String added to the name of the shortened video ('short' by default).

    dest_folder: str, optional
        Folder the video is saved into (by default, same as the original video)

    Returns
    -------
    str
        Full path to the shortened video
    """

    def validate_timestamp(stamp):
        if not isinstance(stamp, str):
            raise ValueError("Timestamp should be a string formatted as hours:minutes:seconds.")
        time = datetime.datetime.strptime(stamp, "%H:%M:%S").time()
        # The above already raises a ValueError if formatting is wrong
        seconds = (time.hour * 60 + time.minute) * 60 + time.second
        if seconds > self.calc_duration():
            raise ValueError("Timestamps must not exceed the video duration.")

    if validate_inputs:
        for stamp in start, end:
            validate_timestamp(stamp)

    output_path = self.make_output_path(suffix, dest_folder)
    command = f'ffmpeg -n -i "{self.video_path}" -ss {start} -to {end} -c:a copy "{output_path}"'
    subprocess.call(command, shell=True)
    return output_path

split 

split(n_splits, suffix='split', dest_folder=None)

Split a video into several shorter ones of equal duration.

Parameters

n_splits : int Number of shorter videos to produce

str, optional

String added to the name of the splits ('short' by default).

str, optional

Folder the video splits are saved into (by default, same as the original video)

Returns

list Paths of the video splits

Source code in deeplabcut/utils/auxfun_videos.py 
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
def split(self, n_splits, suffix="split", dest_folder=None):
    """Split a video into several shorter ones of equal duration.

    Parameters
    ----------
    n_splits : int
        Number of shorter videos to produce

    suffix: str, optional
        String added to the name of the splits ('short' by default).

    dest_folder: str, optional
        Folder the video splits are saved into (by default, same as the original video)

    Returns
    -------
    list
        Paths of the video splits
    """
    if not n_splits > 1:
        raise ValueError("The video should at least be split in half.")
    chunk_dur = self.calc_duration() / n_splits
    splits = np.arange(n_splits + 1) * chunk_dur

    def time_formatter(val):
        return str(datetime.timedelta(seconds=val))

    clips = []
    for n, (start, end) in enumerate(zip(splits, splits[1:], strict=False), start=1):
        clips.append(
            self.shorten(
                time_formatter(start),
                time_formatter(end),
                f"{suffix}{n}",
                dest_folder,
                validate_inputs=False,
            )
        )
    return clips

PlottingSingleFrame 

PlottingSingleFrame(
    clip,
    Dataframe,
    bodyparts2plot,
    tmpfolder,
    index,
    dotsize,
    pcutoff,
    alphavalue,
    colors,
    strwidth=4,
    savelabeled=True,
)

Label frame and save under imagename / this is already cropped (for clip)

Source code in deeplabcut/refine_training_dataset/outlier_frames.py 
 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
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
def PlottingSingleFrame(
    clip,
    Dataframe,
    bodyparts2plot,
    tmpfolder,
    index,
    dotsize,
    pcutoff,
    alphavalue,
    colors,
    strwidth=4,
    savelabeled=True,
):
    """Label frame and save under imagename / this is already cropped (for clip)"""
    from skimage import io

    imagename1 = os.path.join(tmpfolder, "img" + str(index).zfill(strwidth) + ".png")
    imagename2 = os.path.join(tmpfolder, "img" + str(index).zfill(strwidth) + "labeled.png")

    if not os.path.isfile(os.path.join(tmpfolder, "img" + str(index).zfill(strwidth) + ".png")):
        plt.axis("off")
        image = img_as_ubyte(clip.get_frame(index * 1.0 / clip.fps))
        io.imsave(imagename1, image)

        if savelabeled:
            if np.ndim(image) > 2:
                h, w, nc = np.shape(image)
            else:
                h, w = np.shape(image)

            bpts = Dataframe.columns.get_level_values("bodyparts")
            all_bpts = bpts.values[::3]
            df_x, df_y, df_likelihood = Dataframe.values.reshape((Dataframe.shape[0], -1, 3)).T
            bplist = bpts.unique().to_list()
            if Dataframe.columns.nlevels == 3:
                map2bp = list(range(len(all_bpts)))
            else:
                map2bp = [bplist.index(bp) for bp in all_bpts]
            keep = np.flatnonzero(np.isin(all_bpts, bodyparts2plot))

            plt.figure(frameon=False, figsize=(w * 1.0 / 100, h * 1.0 / 100))
            plt.subplots_adjust(left=0, bottom=0, right=1, top=1, wspace=0, hspace=0)
            plt.imshow(image)
            for i, ind in enumerate(keep):
                if df_likelihood[ind, index] > pcutoff:
                    plt.scatter(
                        df_x[ind, index],
                        df_y[ind, index],
                        s=dotsize**2,
                        color=colors(map2bp[i]),
                        alpha=alphavalue,
                    )
            plt.xlim(0, w)
            plt.ylim(0, h)
            plt.axis("off")
            plt.subplots_adjust(left=0, bottom=0, right=1, top=1, wspace=0, hspace=0)
            plt.gca().invert_yaxis()
            plt.savefig(imagename2)
            plt.close("all")

PlottingSingleFramecv2 

PlottingSingleFramecv2(
    cap, Dataframe, bodyparts2plot, tmpfolder, index, dotsize, pcutoff, alphavalue, colors, strwidth=4, savelabeled=True
)

Label frame and save under imagename / cap is not already cropped.

Source code in deeplabcut/refine_training_dataset/outlier_frames.py 
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
1052
1053
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
def PlottingSingleFramecv2(
    cap,
    Dataframe,
    bodyparts2plot,
    tmpfolder,
    index,
    dotsize,
    pcutoff,
    alphavalue,
    colors,
    strwidth=4,
    savelabeled=True,
):
    """Label frame and save under imagename / cap is not already cropped."""
    from skimage import io

    imagename1 = os.path.join(tmpfolder, "img" + str(index).zfill(strwidth) + ".png")
    imagename2 = os.path.join(tmpfolder, "img" + str(index).zfill(strwidth) + "labeled.png")

    if not os.path.isfile(os.path.join(tmpfolder, "img" + str(index).zfill(strwidth) + ".png")):
        plt.axis("off")
        cap.set_to_frame(index)
        frame = cap.read_frame(crop=True)
        if frame is None:
            print("Frame could not be read.")
            return
        image = img_as_ubyte(frame)
        io.imsave(imagename1, image)

        if savelabeled:
            if np.ndim(image) > 2:
                h, w, nc = np.shape(image)
            else:
                h, w = np.shape(image)

            bpts = Dataframe.columns.get_level_values("bodyparts")
            all_bpts = bpts.values[::3]
            df_x, df_y, df_likelihood = Dataframe.values.reshape((Dataframe.shape[0], -1, 3)).T
            bplist = bpts.unique().to_list()
            if Dataframe.columns.nlevels == 3:
                map2bp = list(range(len(all_bpts)))
            else:
                map2bp = [bplist.index(bp) for bp in all_bpts]
            keep = np.flatnonzero(np.isin(all_bpts, bodyparts2plot))

            plt.figure(frameon=False, figsize=(w * 1.0 / 100, h * 1.0 / 100))
            plt.subplots_adjust(left=0, bottom=0, right=1, top=1, wspace=0, hspace=0)
            plt.imshow(image)
            for i, ind in enumerate(keep):
                if df_likelihood[ind, index] > pcutoff:
                    plt.scatter(
                        df_x[ind, index],
                        df_y[ind, index],
                        s=dotsize**2,
                        color=colors(map2bp[i]),
                        alpha=alphavalue,
                    )
            plt.xlim(0, w)
            plt.ylim(0, h)
            plt.axis("off")
            plt.subplots_adjust(left=0, bottom=0, right=1, top=1, wspace=0, hspace=0)
            plt.gca().invert_yaxis()
            plt.savefig(imagename2)
            plt.close("all")

attempt_to_add_video 

attempt_to_add_video(config: str, video: str, copy_videos: bool, coords: list | None) -> bool

Add new videos to the config file at any stage of the project.

Parameters

config : string Full path of the config file in the project.

string

Full path of the video to add to the project.

bool, optional

If this is set to True, the videos will be copied to the project/videos directory. If False, the symlink of the videos will be copied instead. The default is False; if provided it must be either True or False.

list, optional

A list containing the list of cropping coordinates of the video. The default is set to None.

Returns

True iff the video was successfully added to the project

Source code in deeplabcut/refine_training_dataset/outlier_frames.py 
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
def attempt_to_add_video(
    config: str,
    video: str,
    copy_videos: bool,
    coords: list | None,
) -> bool:
    """Add new videos to the config file at any stage of the project.

    Parameters
    ----------
    config : string
        Full path of the config file in the project.

    video : string
        Full path of the video to add to the project.

    copy_videos : bool, optional
        If this is set to True, the videos will be copied to the project/videos directory.
        If False, the symlink of the
        videos will be copied instead. The default is
        ``False``; if provided it must be either ``True`` or ``False``.

    coords: list, optional
        A list containing the list of cropping coordinates of the video. The default is set to None.

    Returns
    -------
    True iff the video was successfully added to the project
    """
    from deeplabcut.create_project import add

    # make sure coords and videos are a list
    videos = [video]
    if coords is not None:
        coords = [coords]

    try:
        add.add_new_videos(config, videos, coords=coords, copy_videos=copy_videos)
    except Exception:
        # can we make a catch here? - in fact we should drop indices from DataCombined
        # if they are in CollectedData.. [ideal behavior; currently pretty unlikely]
        print(
            "AUTOMATIC ADDING OF VIDEO TO CONFIG FILE FAILED! You need to "
            "do this manually for including it in the config.yaml file!"
        )
        print("Videopath:", video, "Coordinates for cropping:", coords)
        return False

    return True

collect_video_paths 

collect_video_paths(
    data_path: str | Path | list[str | Path],
    extensions: str | Sequence[str] | None = None,
    shuffle: bool = False,
    exclude_patterns: Sequence[str] = DEFAULT_EXCLUDE_PATTERNS,
) -> list[Path]

Collects video paths from a given set of data paths: directories, files, or a mix of both. Directories are scanned one level deep (non-recursively).

Files and directories are treated differently with respect to extension filtering: - File paths are accepted as-is when extensions is None; only filtered when extensions is explicitly set. - Directory contents are always filtered by extension: by SUPPORTED_VIDEOS when extensions is None, or by the given value(s) otherwise. - exclude_patterns are always applied to both files and directory contents.

Parameters:

Name Type Description Default

data_path

str | Path | list[str | Path]

Path or list of paths to folders containing videos, or individual video files. Can be a mix of directories and files.

 required

extensions

str | Sequence[str] | None

Controls extension filtering for collected video files. - None (default): file paths are accepted without extension filtering; directories are scanned for files with a recognized video extension. - str or Sequence[str] (e.g. "mp4" or ["mp4", "avi"]): both file paths and directory contents are filtered to only include files matching the given extension(s). - Empty str "" is treated as None (deprecated, keep for backwards compatibility).

 None

shuffle

bool

Whether to shuffle the order of videos. If False, videos are returned in sorted order for deterministic behavior.

 False

exclude_patterns

Sequence[str]

Patterns to exclude from the collection. Defaults to DEFAULT_EXCLUDE_PATTERNS. Set to [] to disable pattern exclusion.

 DEFAULT_EXCLUDE_PATTERNS

Returns:

Type Description
list[Path]

The paths of videos to analyze. Duplicate paths are removed.

Raises:

Type Description
FileNotFoundError

If any path in data_path does not exist.
ValueError

If extensions is an empty sequence.
Source code in deeplabcut/utils/auxfun_videos.py 
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
def collect_video_paths(
    data_path: str | Path | list[str | Path],
    extensions: str | Sequence[str] | None = None,
    shuffle: bool = False,
    exclude_patterns: Sequence[str] = DEFAULT_EXCLUDE_PATTERNS,
) -> list[Path]:
    """
    Collects video paths from a given set of data paths: directories, files, or a mix
    of both. Directories are scanned one level deep (non-recursively).

    Files and directories are treated differently with respect to extension filtering:
    - File paths are accepted as-is when ``extensions`` is ``None``; only filtered when
      ``extensions`` is explicitly set.
    - Directory contents are always filtered by extension: by ``SUPPORTED_VIDEOS`` when
      ``extensions`` is ``None``, or by the given value(s) otherwise.
    - ``exclude_patterns`` are always applied to both files and directory contents.

    Args:
        data_path: Path or list of paths to folders containing videos, or individual
            video files. Can be a mix of directories and files.
        extensions: Controls extension filtering for collected video files.
            - ``None`` (default): file paths are accepted without extension filtering;
              directories are scanned for files with a recognized video extension.
            - ``str`` or ``Sequence[str]`` (e.g. ``"mp4"`` or ``["mp4", "avi"]``):
              both file paths and directory contents are filtered to only include files
              matching the given extension(s).
            - Empty ``str`` ``""`` is treated as ``None`` (deprecated, keep for backwards
              compatibility).
        shuffle: Whether to shuffle the order of videos. If ``False``, videos are
            returned in sorted order for deterministic behavior.
        exclude_patterns: Patterns to exclude from the collection. Defaults to
            ``DEFAULT_EXCLUDE_PATTERNS``. Set to ``[]`` to disable pattern exclusion.

    Returns:
        The paths of videos to analyze. Duplicate paths are removed.

    Raises:
        FileNotFoundError: If any path in ``data_path`` does not exist.
        ValueError: If ``extensions`` is an empty sequence.
    """
    if isinstance(data_path, (str, Path)):
        data_path = [data_path]

    def _coerce_extensions(extensions: str | Sequence[str] | None) -> set[str] | None:
        """Coerce the extensions argument to a set of dot-prefixed suffixes, or None."""
        if extensions is None:
            return None

        if extensions in ["", ("",), [""], {""}]:
            warnings.warn(
                "Passing an empty string for filtering video type extensions is deprecated; pass None instead.",
                DLCDeprecationWarning,
                stacklevel=3,
            )
            return None

        if isinstance(extensions, str):
            return {f".{extensions.lstrip('.').lower()}"}

        if not isinstance(extensions, Sequence):
            raise TypeError(f"extensions must be a string, a sequence or None, got {type(extensions)}")

        if len(extensions) == 0:
            raise ValueError("Video type extensions filter needs to be a non-empty sequence.")
        return {f".{e.lstrip('.').lower()}" for e in extensions}

    explicit_suffixes = _coerce_extensions(extensions)
    implicit_suffixes = {f".{ext.lower()}" for ext in SUPPORTED_VIDEOS}

    videos: list[Path] = []
    for path in map(Path, data_path):
        if not path.exists():
            raise FileNotFoundError(f"Could not find: {path}. Check access rights.")

        if path.is_dir():
            # Discriminate videos from other files; skip excluded patterns (e.g. prior DLC outputs).
            allowed = explicit_suffixes if explicit_suffixes else implicit_suffixes
            videos.extend(
                f
                for f in path.iterdir()
                if f.is_file()
                and f.suffix.lower() in allowed
                and not any(f.match(pattern) for pattern in exclude_patterns)
            )
        elif path.is_file():
            # Accept all caller-supplied files; ONLY filter extensions if set. ALWAYS filter exclude patterns.
            if explicit_suffixes is None or path.suffix.lower() in explicit_suffixes:
                if not any(path.match(pattern) for pattern in exclude_patterns):
                    videos.append(path)

    # Resolve video paths and remove duplicates
    unique_videos = list(dict.fromkeys(v.resolve() for v in videos))
    if shuffle:
        random.shuffle(unique_videos)
    else:
        unique_videos.sort()

    if any(fn.suffix.lower().lstrip(".") not in SUPPORTED_VIDEOS for fn in unique_videos if fn.suffix):
        warnings.warn(
            f"Some videos have unsupported extensions: {unique_videos} \nSupported extensions are: {SUPPORTED_VIDEOS}",
            stacklevel=2,
        )
    return unique_videos

columnwise_spline_interp 

columnwise_spline_interp(data, max_gap=0)

Perform cubic spline interpolation over the columns of data. All gaps of size lower than or equal to max_gap are filled, and data slightly smoothed.

Parameters

data : array_like 2D matrix of data. max_gap : int, optional Maximum gap size to fill. By default, all gaps are interpolated.

Returns

interpolated data with same shape as data

Source code in deeplabcut/post_processing/filtering.py 
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
def columnwise_spline_interp(data, max_gap=0):
    """Perform cubic spline interpolation over the columns of *data*. All gaps of size
    lower than or equal to *max_gap* are filled, and data slightly smoothed.

    Parameters
    ----------
    data : array_like
        2D matrix of data.
    max_gap : int, optional
        Maximum gap size to fill. By default, all gaps are interpolated.

    Returns
    -------
    interpolated data with same shape as *data*
    """
    if np.ndim(data) < 2:
        data = np.expand_dims(data, axis=1)
    nrows, ncols = data.shape
    temp = data.copy()
    valid = ~np.isnan(temp)
    x = np.arange(nrows)
    for i in range(ncols):
        mask = valid[:, i]
        if np.sum(mask) > 3:  # Make sure there are enough points to fit the cubic spline
            spl = CubicSpline(x[mask], temp[mask, i])
            y = spl(x)
            if max_gap > 0:
                inds = np.flatnonzero(np.r_[True, np.diff(mask), True])
                count = np.diff(inds)
                inds = inds[:-1]
                to_fill = np.ones_like(mask)
                for ind, n, is_nan in zip(inds, count, ~mask[inds], strict=False):
                    if is_nan and n > max_gap:
                        to_fill[ind : ind + n] = False
                y[~to_fill] = np.nan
            # Get rid of the interpolation beyond the spline knots
            y[y == 0] = np.nan
            temp[:, i] = y
    return temp

compute_deviations 

compute_deviations(Dataframe, dataname, p_bound, alpha, ARdegree, MAdegree, storeoutput=None)

Fits Seasonal AutoRegressive Integrated Moving Average with eXogenous regressors model to data and computes confidence interval as well as mean fit.

Source code in deeplabcut/refine_training_dataset/outlier_frames.py 
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
def compute_deviations(Dataframe, dataname, p_bound, alpha, ARdegree, MAdegree, storeoutput=None):
    """Fits Seasonal AutoRegressive Integrated Moving Average with eXogenous regressors
    model to data and computes confidence interval as well as mean fit."""

    print("Fitting state-space models with parameters:", ARdegree, MAdegree)
    df_x, df_y, df_likelihood = Dataframe.values.reshape((Dataframe.shape[0], -1, 3)).T
    preds = []
    for row in range(len(df_x)):
        x = df_x[row]
        y = df_y[row]
        p = df_likelihood[row]
        meanx, CIx = FitSARIMAXModel(x, p, p_bound, alpha, ARdegree, MAdegree)
        meany, CIy = FitSARIMAXModel(y, p, p_bound, alpha, ARdegree, MAdegree)
        distance = np.sqrt((x - meanx) ** 2 + (y - meany) ** 2)
        significant = (x < CIx[:, 0]) + (x > CIx[:, 1]) + (y < CIy[:, 0]) + (y > CIy[:, 1])
        preds.append(np.c_[distance, significant, meanx, meany, CIx, CIy])

    columns = Dataframe.columns
    # Use the existing valid keypoint combinations, in their original order.
    # The goal is to extract each stream (e.g. Scorer/ID/Bodypart) as a separate column,
    # and then build, for each stat, a MultiIndex with the same levels, i.e.
    # Scorer/ID/Bodypart/stat (see stats below).
    # Note, this could be built from "y" as well without any difference in the output
    base_cols = Dataframe.xs("x", axis=1, level="coords", drop_level=True).columns
    stats = [
        "distance",
        "sig",
        "meanx",
        "meany",
        "lowerCIx",
        "higherCIx",
        "lowerCIy",
        "higherCIy",
    ]
    pdindex = pd.MultiIndex.from_tuples(
        [(*col, stat) for col in base_cols for stat in stats],
        names=[n for n in columns.names if n != "coords"] + ["stats"],
    )
    data = pd.DataFrame(np.concatenate(preds, axis=1), columns=pdindex)  # preds (n_frames, n_stats * n_streams)
    # average distance and average # significant differences avg. over comparisonbodyparts
    d = data.xs("distance", axis=1, level=-1).mean(axis=1).values
    o = data.xs("sig", axis=1, level=-1).mean(axis=1).values

    if storeoutput == "full":
        data.to_hdf(
            dataname.split(".h5")[0] + "filtered.h5",
            key="df_with_missing",
            format="table",
            mode="w",
        )
        return d, o, data
    else:
        return d, o

convertparms2start 

convertparms2start(pn)

Creating a start value for sarimax in case of an value error See: https://groups.google.com/forum/#!topic/pystatsmodels/S_Fo53F25Rk

Source code in deeplabcut/refine_training_dataset/outlier_frames.py 
552
553
554
555
556
557
558
559
560
561
562
def convertparms2start(pn):
    """Creating a start value for sarimax in case of an value error
    See: https://groups.google.com/forum/#!topic/pystatsmodels/S_Fo53F25Rk"""
    if "ar." in pn:
        return 0
    elif "ma." in pn:
        return 0
    elif "sigma" in pn:
        return 1
    else:
        return 0

extract_outlier_frames 

extract_outlier_frames(
    config,
    videos,
    video_extensions: str | Sequence[str] | None = None,
    shuffle=1,
    trainingsetindex=0,
    outlieralgorithm="jump",
    frames2use=None,
    comparisonbodyparts="all",
    epsilon=20,
    p_bound=0.01,
    ARdegree=3,
    MAdegree=1,
    alpha=0.01,
    extractionalgorithm="kmeans",
    automatic=False,
    cluster_resizewidth=30,
    cluster_color=False,
    opencv=True,
    savelabeled=False,
    copy_videos=False,
    destfolder=None,
    modelprefix="",
    track_method="",
    **kwargs
)

Extracts the outlier frames.

Extracts the outlier frames if the predictions are not correct for a certain video from the cropped video running from start to stop as defined in config.yaml.

Another crucial parameter in config.yaml is how many frames to extract numframes2extract.

Parameters

config: str Full path of the config.yaml file.

list[str]

The full paths to videos for analysis or a path to the directory, where all the videos with same extension are stored.

str | Sequence[str] | None, optional, default=None

Controls how videos are filtered, based on file extension. File paths and directory contents are treated differently: - None (default): file paths are accepted as-is; directories are scanned for files with a recognized video extension. - str or Sequence[str] (e.g. "mp4" or ["mp4", "avi"]): both file paths and directory contents are filtered by the given extension(s).

int, optional, default=1

The shuffle index of training dataset. The extracted frames will be stored in the labeled-dataset for the corresponding shuffle of training dataset.

int, optional, default=0

Integer specifying which TrainingsetFraction to use. Note that TrainingFraction is a list in config.yaml.

str, optional, default="jump".

String specifying the algorithm used to detect the outliers.

  • 'fitting' fits an Auto Regressive Integrated Moving Average model to the data and computes the distance to the estimated data. Larger distances than epsilon are then potentially identified as outliers
  • 'jump' identifies larger jumps than 'epsilon' in any body part
  • 'uncertain' looks for frames with confidence below p_bound
  • 'manual' launches a GUI from which the user can choose the frames
  • 'list' looks for user to provide a list of frame numbers to use, 'frames2use'. In this case, 'extractionalgorithm' is forced to be 'uniform.'
list[str], optional, default=None

If 'outlieralgorithm' is 'list', provide the list of frames here.

list[str] or str, optional, default="all"

This selects the body parts for which the comparisons with the outliers are carried out. If "all", then all body parts from config.yaml are used. If a list of strings that are a subset of the full list E.g. ['hand','Joystick'] for the demo Reaching-Mackenzie-2018-08-30/config.yaml to select only these body parts.

float between 0 and 1, optional, default=0.01

For outlieralgorithm 'uncertain' this parameter defines the likelihood below which a body part will be flagged as a putative outlier.

float, optional, default=20

If 'outlieralgorithm' is 'fitting', this is the float bound according to which frames are picked when the (average) body part estimate deviates from model fit.

If 'outlieralgorithm' is 'jump', this is the float bound specifying the distance by which body points jump from one frame to next (Euclidean distance).

int, optional, default=3

For outlieralgorithm 'fitting': Autoregressive degree of ARIMA model degree. (Note we use SARIMAX without exogeneous and seasonal part) See https://www.statsmodels.org/dev/generated/statsmodels.tsa.statespace.sarimax.SARIMAX.html

int, optional, default=1

For outlieralgorithm 'fitting': Moving Average degree of ARIMA model degree. (Note we use SARIMAX without exogeneous and seasonal part) See https://www.statsmodels.org/dev/generated/statsmodels.tsa.statespace.sarimax.SARIMAX.html

float, optional, default=0.01

Significance level for detecting outliers based on confidence interval of fitted ARIMA model. Only the distance is used however.

str, optional, default="kmeans"

String specifying the algorithm to use for selecting the frames from the identified putatative outlier frames. Currently, deeplabcut supports either kmeans or uniform based selection (same logic as for extract_frames).

bool, optional, default=False

If True, extract outliers without being asked for user feedback.

number, default=30

If "extractionalgorithm" is "kmeans", one can change the width to which the images are downsampled (aspect ratio is fixed).

bool, optional, default=False

If False, each downsampled image is treated as a grayscale vector (discarding color information). If True, then the color channels are considered. This increases the computational complexity.

bool, optional, default=True

Uses openCV for loading & extractiong (otherwise moviepy (legacy)).

bool, optional, default=False

If True, frame are saved with predicted labels in each folder.

bool, optional, default=False

If True, newly-added videos (from which outlier frames are extracted) are copied to the project folder. By default, symbolic links are created instead.

str or None, optional, default=None

Specifies the destination folder that was used for storing analysis data. If None, the path of the video is used.

str, optional, default=""

Directory containing the deeplabcut models to use when evaluating the network. By default, the models are assumed to exist in the project folder.

str, optional, default=""

Specifies the tracker used to generate the data. Empty by default (corresponding to a single animal project). For multiple animals, must be either 'box', 'skeleton', or 'ellipse' and will be taken from the config.yaml file if none is given.

additional arguments.

For torch-based shuffles, can be used to specify: - snapshot_index - detector_snapshot_index

Returns

None

Examples

Extract the frames with default settings on Windows.

deeplabcut.extract_outlier_frames( 'C:\myproject\reaching-task\config.yaml', ['C:\yourusername\rig-95\Videos\reachingvideo1.avi'], )

Extract the frames with default settings on Linux/MacOS.

deeplabcut.extract_outlier_frames( '/analysis/project/reaching-task/config.yaml', ['/analysis/project/video/reachinvideo1.avi'], )

Extract the frames using the "kmeans" algorithm.

deeplabcut.extract_outlier_frames( '/analysis/project/reaching-task/config.yaml', ['/analysis/project/video/reachinvideo1.avi'], extractionalgorithm='kmeans', )

Extract the frames using the "kmeans" algorithm and "epsilon=5" pixels.

deeplabcut.extract_outlier_frames( '/analysis/project/reaching-task/config.yaml', ['/analysis/project/video/reachinvideo1.avi'], epsilon=5, extractionalgorithm='kmeans', )

Source code in deeplabcut/refine_training_dataset/outlier_frames.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
302
303
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
399
400
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
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
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
@renamed_parameter(old="videotype", new="video_extensions", since="3.0.0")
def extract_outlier_frames(
    config,
    videos,
    video_extensions: str | Sequence[str] | None = None,
    shuffle=1,
    trainingsetindex=0,
    outlieralgorithm="jump",
    frames2use=None,
    comparisonbodyparts="all",
    epsilon=20,
    p_bound=0.01,
    ARdegree=3,
    MAdegree=1,
    alpha=0.01,
    extractionalgorithm="kmeans",
    automatic=False,
    cluster_resizewidth=30,
    cluster_color=False,
    opencv=True,
    savelabeled=False,
    copy_videos=False,
    destfolder=None,
    modelprefix="",
    track_method="",
    **kwargs,
):
    """Extracts the outlier frames.

    Extracts the outlier frames if the predictions are not correct for a certain video
    from the cropped video running from start to stop as defined in config.yaml.

    Another crucial parameter in config.yaml is how many frames to extract
    ``numframes2extract``.

    Parameters
    ----------
    config: str
        Full path of the config.yaml file.

    videos : list[str]
        The full paths to videos for analysis or a path to the directory, where all the
        videos with same extension are stored.

    video_extensions : str | Sequence[str] | None, optional, default=None
        Controls how ``videos`` are filtered, based on file extension.
        File paths and directory contents are treated differently:
        - ``None`` (default): file paths are accepted as-is; directories are
          scanned for files with a recognized video extension.
        - ``str`` or ``Sequence[str]`` (e.g. ``"mp4"`` or ``["mp4", "avi"]``):
          both file paths and directory contents are filtered by the given
          extension(s).

    shuffle : int, optional, default=1
        The shuffle index of training dataset. The extracted frames will be stored in
        the labeled-dataset for the corresponding shuffle of training dataset.

    trainingsetindex: int, optional, default=0
        Integer specifying which TrainingsetFraction to use.
        Note that TrainingFraction is a list in config.yaml.

    outlieralgorithm: str, optional, default="jump".
        String specifying the algorithm used to detect the outliers.

        * ``'fitting'`` fits an Auto Regressive Integrated Moving Average model to the
          data and computes the distance to the estimated data. Larger distances than
          epsilon are then potentially identified as outliers
        * ``'jump'`` identifies larger jumps than 'epsilon' in any body part
        * ``'uncertain'`` looks for frames with confidence below p_bound
        * ``'manual'`` launches a GUI from which the user can choose the frames
        * ``'list'`` looks for user to provide a list of
          frame numbers to use, 'frames2use'.
          In this case, ``'extractionalgorithm'`` is forced to be ``'uniform.'``

    frames2use: list[str], optional, default=None
        If ``'outlieralgorithm'`` is ``'list'``, provide the list of frames here.

    comparisonbodyparts: list[str] or str, optional, default="all"
        This selects the body parts for which the comparisons with the outliers are
        carried out. If ``"all"``, then all body parts from config.yaml are used. If a
        list of strings that are a subset of the full list E.g. ['hand','Joystick'] for
        the demo Reaching-Mackenzie-2018-08-30/config.yaml to select only these body
        parts.

    p_bound: float between 0 and 1, optional, default=0.01
        For outlieralgorithm ``'uncertain'`` this parameter defines the likelihood
        below which a body part will be flagged as a putative outlier.

    epsilon: float, optional, default=20
        If ``'outlieralgorithm'`` is ``'fitting'``, this is the float bound according
        to which frames are picked when the (average) body part estimate deviates from
        model fit.

        If ``'outlieralgorithm'`` is ``'jump'``, this is the float bound specifying the
        distance by which body points jump from one frame to next (Euclidean distance).

    ARdegree: int, optional, default=3
        For outlieralgorithm ``'fitting'``: Autoregressive degree of ARIMA model degree.
        (Note we use SARIMAX without exogeneous and seasonal part)
        See https://www.statsmodels.org/dev/generated/statsmodels.tsa.statespace.sarimax.SARIMAX.html

    MAdegree: int, optional, default=1
        For outlieralgorithm ``'fitting'``: Moving Average degree of ARIMA model degree.
        (Note we use SARIMAX without exogeneous and seasonal part)
        See https://www.statsmodels.org/dev/generated/statsmodels.tsa.statespace.sarimax.SARIMAX.html

    alpha: float, optional, default=0.01
        Significance level for detecting outliers based on confidence interval of
        fitted ARIMA model. Only the distance is used however.

    extractionalgorithm : str, optional, default="kmeans"
        String specifying the algorithm to use for selecting the frames from the
        identified putatative outlier frames. Currently, deeplabcut supports either
        ``kmeans`` or ``uniform`` based selection (same logic as for extract_frames).

    automatic : bool, optional, default=False
        If ``True``, extract outliers without being asked for user feedback.

    cluster_resizewidth: number, default=30
        If ``"extractionalgorithm"`` is ``"kmeans"``, one can change the width to which
        the images are downsampled (aspect ratio is fixed).

    cluster_color: bool, optional, default=False
        If ``False``, each downsampled image is treated as a grayscale vector
        (discarding color information). If ``True``, then the color channels are
        considered. This increases the computational complexity.

    opencv: bool, optional, default=True
        Uses openCV for loading & extractiong (otherwise moviepy (legacy)).

    savelabeled: bool, optional, default=False
        If ``True``, frame are saved with predicted labels in each folder.

    copy_videos: bool, optional, default=False
        If True, newly-added videos (from which outlier frames are extracted) are
        copied to the project folder. By default, symbolic links are created instead.

    destfolder: str or None, optional, default=None
        Specifies the destination folder that was used for storing analysis data. If
        ``None``, the path of the video is used.

    modelprefix: str, optional, default=""
        Directory containing the deeplabcut models to use when evaluating the network.
        By default, the models are assumed to exist in the project folder.

    track_method: str, optional, default=""
         Specifies the tracker used to generate the data.
         Empty by default (corresponding to a single animal project).
         For multiple animals, must be either 'box', 'skeleton', or 'ellipse' and will
         be taken from the config.yaml file if none is given.

    kwargs: additional arguments.
        For torch-based shuffles, can be used to specify:
            - snapshot_index
            - detector_snapshot_index

    Returns
    -------
    None

    Examples
    --------

    Extract the frames with default settings on Windows.

    >>> deeplabcut.extract_outlier_frames(
            'C:\\myproject\\reaching-task\\config.yaml',
            ['C:\\yourusername\\rig-95\\Videos\\reachingvideo1.avi'],
        )

    Extract the frames with default settings on Linux/MacOS.

    >>> deeplabcut.extract_outlier_frames(
            '/analysis/project/reaching-task/config.yaml',
            ['/analysis/project/video/reachinvideo1.avi'],
        )

    Extract the frames using the "kmeans" algorithm.

    >>> deeplabcut.extract_outlier_frames(
            '/analysis/project/reaching-task/config.yaml',
            ['/analysis/project/video/reachinvideo1.avi'],
            extractionalgorithm='kmeans',
        )

    Extract the frames using the "kmeans" algorithm and ``"epsilon=5"`` pixels.

    >>> deeplabcut.extract_outlier_frames(
            '/analysis/project/reaching-task/config.yaml',
            ['/analysis/project/video/reachinvideo1.avi'],
            epsilon=5,
            extractionalgorithm='kmeans',
        )
    """

    cfg = auxiliaryfunctions.read_config(config)
    bodyparts = auxiliaryfunctions.intersection_of_body_parts_and_ones_given_by_user(cfg, comparisonbodyparts)
    if not len(bodyparts):
        raise ValueError("No valid bodyparts were selected.")

    track_method = auxfun_multianimal.get_track_method(cfg, track_method=track_method)

    DLCscorer, DLCscorerlegacy = auxiliaryfunctions.get_scorer_name(
        cfg,
        shuffle,
        trainFraction=cfg["TrainingFraction"][trainingsetindex],
        modelprefix=modelprefix,
        **kwargs,
    )

    Videos = collect_video_paths(videos, extensions=video_extensions)
    if len(Videos) == 0:
        print("No suitable videos found in", videos)

    for video in Videos:
        if destfolder is None:
            videofolder = str(Path(video).parents[0])
        else:
            videofolder = destfolder
        vname = os.path.splitext(os.path.basename(video))[0]

        try:
            df, dataname, _, _ = auxiliaryfunctions.load_analyzed_data(
                videofolder, vname, DLCscorer, track_method=track_method
            )
            metadata = auxiliaryfunctions.load_video_metadata(videofolder, vname, DLCscorer)
            nframes = len(df)
            startindex = max([int(np.floor(nframes * cfg["start"])), 0])
            stopindex = min([int(np.ceil(nframes * cfg["stop"])), nframes])
            Index = np.arange(stopindex - startindex) + startindex

            # offset if the data was cropped
            # note: When output video is also cropped, the keypoints should be shifted back.
            out_x1, out_y1 = _read_video_specific_cropping_margins(config, video)
            if metadata.get("data", {}).get("cropping"):
                x1, _, y1, _ = metadata["data"]["cropping_parameters"]
                df.iloc[:, df.columns.get_level_values(level="coords") == "x"] += x1 - out_x1
                df.iloc[:, df.columns.get_level_values(level="coords") == "y"] += y1 - out_y1

            df = df.iloc[Index]
            mask = df.columns.get_level_values("bodyparts").isin(bodyparts)
            df_temp = df.loc[:, mask]
            Indices = []
            if outlieralgorithm == "uncertain":
                p = df_temp.xs("likelihood", level="coords", axis=1)
                ind = df_temp.index[(p < p_bound).any(axis=1)].tolist()
                Indices.extend(ind)
            elif outlieralgorithm == "jump":
                temp_dt = df_temp.diff(axis=0) ** 2
                temp_dt.drop("likelihood", axis=1, level="coords", inplace=True)
                sum_ = temp_dt.groupby(level="bodyparts", axis=1).sum()
                ind = df_temp.index[(sum_ > epsilon**2).any(axis=1)].tolist()
                Indices.extend(ind)
            elif outlieralgorithm == "fitting":
                d, o = compute_deviations(df_temp, dataname, p_bound, alpha, ARdegree, MAdegree)
                # Some heuristics for extracting frames based on distance:
                ind = np.flatnonzero(d > epsilon)  # time points with at least average difference of epsilon
                if (
                    len(ind) < cfg["numframes2pick"] * 2 and len(d) > cfg["numframes2pick"] * 2
                ):  # if too few points qualify, extract the most distant ones.
                    ind = np.argsort(d)[::-1][: cfg["numframes2pick"] * 2]
                Indices.extend(ind)
            elif outlieralgorithm == "manual":
                from deeplabcut.gui.widgets import launch_napari

                added_video = attempt_to_add_video(
                    config=config,
                    video=video,
                    copy_videos=copy_videos,
                    coords=None,
                )
                if added_video:
                    project_video_path = Path(cfg["project_path"]) / "videos" / Path(video).name
                    _ = launch_napari([project_video_path, dataname])
                return

            elif outlieralgorithm == "list":
                if frames2use is not None:
                    try:
                        frames2use = np.array(frames2use).astype("int")
                    except ValueError():
                        print(
                            "Could not cast frames2use into np array, "
                            "please check that frames2use is a simply a list of integers!"
                        )
                        raise
                    Indices.extend(frames2use)
                else:
                    raise ValueError('Expected list of frames2use for outlieralgorithm "list"!')
            else:
                raise ValueError(f"outlieralgorithm {outlieralgorithm} not recognized!")

            # Run always except when the outlieralgorithm == manual.
            if not outlieralgorithm == "manual":
                Indices = np.sort(list(set(Indices)))  # remove repetitions.
                print(
                    "Method ",
                    outlieralgorithm,
                    " found ",
                    len(Indices),
                    " putative outlier frames.",
                )
                print(
                    "Do you want to proceed with extracting ",
                    cfg["numframes2pick"],
                    " of those?",
                )
                if outlieralgorithm == "uncertain" or outlieralgorithm == "jump":
                    print(
                        "If this list is very large, perhaps consider changing the parameters "
                        "(start, stop, p_bound, comparisonbodyparts) or use a different method."
                    )
                elif outlieralgorithm == "fitting":
                    print(
                        "If this list is very large, perhaps consider changing the parameters "
                        "(start, stop, epsilon, ARdegree, MAdegree, alpha, comparisonbodyparts) "
                        "or use a different method."
                    )

                if not automatic:
                    askuser = input("yes/no")
                else:
                    askuser = "Ja"

                if askuser == "y" or askuser == "yes" or askuser == "Ja" or askuser == "ha":  # multilanguage support :)
                    # Now extract from those Indices!
                    ExtractFramesbasedonPreselection(
                        Indices,
                        extractionalgorithm,
                        df,
                        video,
                        cfg,
                        config,
                        opencv,
                        cluster_resizewidth,
                        cluster_color,
                        savelabeled,
                        copy_videos=copy_videos,
                    )
                else:
                    print("Nothing extracted, please change the parameters and start again...")
        except FileNotFoundError as e:
            print(e)
            print(
                "It seems the video has not been analyzed yet, or the video is not found! "
                "You can only refine the labels after the a video is analyzed. "
                "Please run 'analyze_video' first. "
                "Or, please double check your video file path"
            )

find_outliers_in_raw_data 

find_outliers_in_raw_data(
    config,
    pickle_file,
    video_file,
    pcutoff=0.1,
    percentiles=(5, 95),
    with_annotations=True,
    extraction_algo="kmeans",
    copy_videos=False,
)

Extract outlier frames from either raw detections or assemblies of multiple animals.

Parameter

config : str Absolute path to the project config.yaml.

str

Path to a _full.pickle or _assemblies.pickle.

str

Path to the corresponding video file for frame extraction.

float, optional (default=0.1)

Detection confidence threshold below which frames are flagged as containing outliers. Only considered if raw detections are passed in.

tuple, optional (default=(5, 95))

Assemblies are considered outliers if their areas are beyond the 5th and 95th percentiles. Must contain a lower and upper bound.

bool, optional (default=True)

If true, extract frames and the corresponding network predictions. Otherwise, only the frames are extracted.

string, optional (default="kmeans")

Outlier detection algorithm. Must be either uniform or kmeans.

bool, optional (default=False)

If True, newly-added videos (from which outlier frames are extracted) are copied to the project folder. By default, symbolic links are created instead.

Source code in deeplabcut/refine_training_dataset/outlier_frames.py 
 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
def find_outliers_in_raw_data(
    config,
    pickle_file,
    video_file,
    pcutoff=0.1,
    percentiles=(5, 95),
    with_annotations=True,
    extraction_algo="kmeans",
    copy_videos=False,
):
    """Extract outlier frames from either raw detections or assemblies of multiple
    animals.

    Parameter
    ----------
    config : str
        Absolute path to the project config.yaml.

    pickled_file : str
        Path to a *_full.pickle or *_assemblies.pickle.

    video_file : str
        Path to the corresponding video file for frame extraction.

    pcutoff : float, optional (default=0.1)
        Detection confidence threshold below which frames are flagged as
        containing outliers. Only considered if raw detections are passed in.

    percentiles : tuple, optional (default=(5, 95))
        Assemblies are considered outliers if their areas are beyond the 5th
        and 95th percentiles. Must contain a lower and upper bound.

    with_annotations : bool, optional (default=True)
        If true, extract frames and the corresponding network predictions.
        Otherwise, only the frames are extracted.

    extraction_algo : string, optional (default="kmeans")
        Outlier detection algorithm. Must be either ``uniform`` or ``kmeans``.

    copy_videos : bool, optional (default=False)
        If True, newly-added videos (from which outlier frames are extracted) are
        copied to the project folder. By default, symbolic links are created instead.
    """
    if extraction_algo not in ("kmeans", "uniform"):
        raise ValueError(f"Unsupported extraction algorithm {extraction_algo}.")

    video_name = Path(video_file).stem
    pickle_name = Path(pickle_file).stem
    if not pickle_name.startswith(video_name):
        raise ValueError("Video and pickle files do not match.")

    with open(pickle_file, "rb") as file:
        data = pickle.load(file)
    if pickle_file.endswith("_full.pickle"):
        inds, data = find_outliers_in_raw_detections(data, threshold=pcutoff)
        with_annotations = False
    elif pickle_file.endswith("_assemblies.pickle"):
        assemblies = dict()
        for k, lst in data.items():
            if k == "single":
                continue
            ass = []
            for vals in lst:
                a = inferenceutils.Assembly(len(vals))
                a.data = vals
                ass.append(a)
            assemblies[k] = ass
        inds = inferenceutils.find_outlier_assemblies(assemblies, qs=percentiles)
    else:
        raise OSError(f"Raw data file {pickle_file} could not be parsed.")

    cfg = auxiliaryfunctions.read_config(config)
    ExtractFramesbasedonPreselection(
        inds,
        extraction_algo,
        data,
        video=video_file,
        cfg=cfg,
        config=config,
        savelabeled=False,
        with_annotations=with_annotations,
        copy_videos=copy_videos,
    )

find_outliers_in_raw_detections 

find_outliers_in_raw_detections(pickled_data, algo='uncertain', threshold=0.1, kept_keypoints=None)

Find outlier frames from the raw detections of multiple animals.

Parameter

pickled_data : dict Data in the *_full.pickle file obtained after analyze_videos.

string, optional (default="uncertain")

Outlier detection algorithm. Currently, only 'uncertain' is supported for multi-animal raw detections.

float, optional (default=0.1)

Detection confidence threshold below which frames are flagged as containing outliers. Only considered if algo==uncertain.

list, optional (default=None)

Indices in the list of labeled body parts to be kept of the analysis. By default, all keypoints are used for outlier search.

Returns

candidates : list Indices of video frames containing potential outliers

Source code in deeplabcut/refine_training_dataset/outlier_frames.py 
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
def find_outliers_in_raw_detections(pickled_data, algo="uncertain", threshold=0.1, kept_keypoints=None):
    """Find outlier frames from the raw detections of multiple animals.

    Parameter
    ----------
    pickled_data : dict
        Data in the *_full.pickle file obtained after `analyze_videos`.

    algo : string, optional (default="uncertain")
        Outlier detection algorithm. Currently, only 'uncertain' is supported
        for multi-animal raw detections.

    threshold: float, optional (default=0.1)
        Detection confidence threshold below which frames are flagged as
        containing outliers. Only considered if `algo`==`uncertain`.

    kept_keypoints : list, optional (default=None)
        Indices in the list of labeled body parts to be kept of the analysis.
        By default, all keypoints are used for outlier search.

    Returns
    -------
    candidates : list
        Indices of video frames containing potential outliers
    """
    if algo != "uncertain":
        raise ValueError("Only method 'uncertain' is currently supported.")

    try:
        _ = pickled_data.pop("metadata")
    except KeyError:
        pass

    def get_frame_ind(s):
        return int(re.findall(r"\d+", s)[0])

    candidates = []
    data = dict()
    for frame_name, dict_ in pickled_data.items():
        frame_ind = get_frame_ind(frame_name)
        temp_coords = dict_["coordinates"][0]
        temp = dict_["confidence"]
        if kept_keypoints is not None:
            temp_coords = [temp_coords[i] for i in kept_keypoints]
            temp = [temp[i] for i in kept_keypoints]
        coords = np.concatenate(temp_coords)
        conf = np.concatenate(temp)
        data[frame_ind] = np.c_[coords, conf].squeeze()
        if np.any(conf < threshold):
            candidates.append(frame_ind)
    return candidates, data

merge_datasets 

merge_datasets(config, forceiterate=None)

Merge the original training dataset with the newly refined data.

Checks if the original training dataset can be merged with the newly refined training dataset. To do so it will check if the frames in all extracted video sets were relabeled.

If this is the case then the "iteration" variable is advanced by 1.

Parameters

config: str Full path of the config.yaml file.

int or None, optional, default=None

If an integer is given the iteration variable is set to this value This is only done if all datasets were labeled or refined.

Examples

deeplabcut.merge_datasets('/analysis/project/reaching-task/config.yaml')

Source code in deeplabcut/refine_training_dataset/outlier_frames.py 
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
def merge_datasets(config, forceiterate=None):
    """Merge the original training dataset with the newly refined data.

    Checks if the original training dataset can be merged with the newly refined
    training dataset. To do so it will check if the frames in all extracted video sets
    were relabeled.

    If this is the case then the ``"iteration"`` variable is advanced by 1.

    Parameters
    ----------
    config: str
        Full path of the config.yaml file.

    forceiterate: int or None, optional, default=None
        If an integer is given the iteration variable is set to this value
        This is only done if all datasets were labeled or refined.

    Examples
    --------

    >>> deeplabcut.merge_datasets('/analysis/project/reaching-task/config.yaml')
    """

    cfg = auxiliaryfunctions.read_config(config)
    config_path = Path(config).parents[0]

    bf = Path(str(config_path / "labeled-data"))
    allfolders = [
        os.path.join(bf, fn) for fn in os.listdir(bf) if "_labeled" not in fn and not fn.startswith(".")
    ]  # exclude labeled data folders and temporary files
    flagged = False
    for _findex, folder in enumerate(allfolders):
        if os.path.isfile(os.path.join(folder, "MachineLabelsRefine.h5")):  # Folder that was manually refine...
            pass
        elif os.path.isfile(
            os.path.join(folder, "CollectedData_" + cfg["scorer"] + ".h5")
        ):  # Folder that contains human data set...
            pass
        else:
            print("The following folder was not manually refined,...", folder)
            flagged = True
            pass  # this folder does not contain a MachineLabelsRefine file (not updated...)

    if not flagged:
        # updates iteration by 1
        iter_prev = cfg["iteration"]
        if not forceiterate:
            cfg["iteration"] = int(iter_prev + 1)
        else:
            cfg["iteration"] = forceiterate

        auxiliaryfunctions.write_config(config, cfg)

        print("Merged data sets and updated refinement iteration to " + str(cfg["iteration"]) + ".")
        print("Now you can create a new training set for the expanded annotated images (use create_training_dataset).")
    else:
        print("Please label, or remove the un-corrected folders.")

renamed_parameter 

renamed_parameter(*, old: str, new: str, since: str | None = None) -> Callable[[Callable[P, R]], Callable[P, R]]

Support a renamed keyword argument while warning callers to update.

Parameters:

Name Type Description Default

old

str

The old parameter name that callers may still pass.

 required

new

str

The current parameter name the function actually accepts.

 required

since

str | None

Version when the rename happened.

 None
Rules
  • new must be the name used in the function signature and all internal call-sites. old must not appear in the signature.
  • Do not chain renames. If A was renamed to B and B is later renamed to C, replace the A→B decorator with A→C directly rather than stacking a second decorator. Example: @renamed_parameter(old="A", new="C", since="12.4.0") @renamed_parameter(old="B", new="C", since="13.0.0") def func(*, C: int): print(f"C={C}")
  • Multiple independent renames on the same function (e.g. batchsize→batch_size and videotype→video_extensions) are fine as long as they do not form a chain.
  • This decorator only intercepts keyword arguments. Positional arguments are passed through unchanged; renaming a parameter that callers commonly pass positionally will not be caught.
Source code in deeplabcut/utils/deprecation.py 
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
def renamed_parameter(
    *,
    old: str,
    new: str,
    since: str | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Support a renamed keyword argument while warning callers to update.

    Args:
        old: The old parameter name that callers may still pass.
        new: The current parameter name the function actually accepts.
        since: Version when the rename happened.

    Rules:
        - ``new`` must be the name used in the function signature and all
          internal call-sites.  ``old`` must **not** appear in the signature.
        - Do **not** chain renames.  If ``A`` was renamed to ``B`` and ``B``
          is later renamed to ``C``, replace the ``A→B`` decorator with
          ``A→C`` directly rather than stacking a second decorator.
            Example:
                @renamed_parameter(old="A", new="C", since="12.4.0")
                @renamed_parameter(old="B", new="C", since="13.0.0")
                def func(*, C: int):
                    print(f"C={C}")
        - Multiple independent renames on the same function (e.g.
          ``batchsize→batch_size`` *and* ``videotype→video_extensions``) are fine
          as long as they do not form a chain.
        - This decorator only intercepts **keyword** arguments.  Positional
          arguments are passed through unchanged; renaming a parameter that
          callers commonly pass positionally will not be caught.
    """

    def decorator(fn: Callable[P, R]) -> Callable[P, R]:
        sig = inspect.signature(fn)

        # Guard: disallow chaining renames (A→B stacked on top of B→C).
        existing = getattr(fn, "__deprecated_params__", ())
        for prev in existing:
            if prev.old_parameter == new:
                raise ValueError(
                    f"@renamed_parameter: chaining renames is not allowed. "
                    f"'{old}' → '{new}' would chain with the existing "
                    f"'{prev.old_parameter}' → '{prev.new_parameter}' rename "
                    f"on {fn.__qualname__}. "
                    f"Use '{old}' → '{prev.new_parameter}' directly instead."
                )

        # Guard: 'new' must actually exist in the function's signature.
        if new not in sig.parameters:
            raise ValueError(
                f"@renamed_parameter: '{new}' is not a parameter of "
                f"{fn.__qualname__}. "
                f"Available parameters: {list(sig.parameters)}"
            )

        # Guard: 'old' must NOT exist in the signature.
        if old in sig.parameters:
            raise ValueError(
                f"@renamed_parameter: '{old}' is still a parameter of "
                f"{fn.__qualname__}. Use either old name or new name: '{new}'."
            )

        info = DeprecationInfo(
            kind="parameter",
            target=fn.__qualname__,
            since=since,
            old_parameter=old,
            new_parameter=new,
        )
        message = info.format_message()

        @functools.wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            if old in kwargs:
                if new in kwargs:
                    raise TypeError(f"{fn.__qualname__} received both '{old}' and '{new}'. Use only '{new}'.")
                warnings.warn(message, DLCDeprecationWarning, stacklevel=2)
                kwargs[new] = kwargs.pop(old)
            return fn(*args, **kwargs)

        wrapper.__deprecated_params__ = (*existing, info)
        return wrapper

    return decorator