Skip to content

utils

This section contains the documentation of movinets_helper\utils.py.

Help functions.

create_class_map(path)

Given a path to a labels.txt file, creates a class map.

Helper function to obtain the classes for the labels.

Parameters:

Name Type Description Default
path str or Path

Path pointing to the labels.txt file.

 required

Returns:

Type Description
Dict[str, int]

Dict[str, int]: Dict mapping from labels to classes.
Source code in movinets_helper/utils.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def create_class_map(path: Union[str, Path]) -> Dict[str, int]:
    """Given a path to a labels.txt file, creates a class map.

    Helper function to obtain the classes for the labels.

    Args:
        path (str or Path): Path pointing to the labels.txt file.

    Returns:
        Dict[str, int]: Dict mapping from labels to classes.
    """
    if isinstance(path, str):
        path = Path(path)
    return {l: i for i, l in enumerate(path.read_text().split("\n"))}

get_chunks(l, n)

Yield successive n-sized chunks from l.

Used to create n sublists from a list l. copied from: https://github.com/ferreirafabio/video2tfrecord/blob/7aa2c6312e2bc97baed7386b8c92c591769ee5bb/video2tfrecord.py#L55

Parameters:

Name Type Description Default
l List[Union[str, Path]] required
n int required

Returns:

Type Description
Iterable[Union[str, Path]]

Iterable[Union[str, Path]]:
Example

filenames_split = list(get_chunks(filenames, n_videos_in_record))

To obtain the files from a dataframe.

next(get_chunks(dataset_df[["classes", "files"]].to_records(index=False), 10))

Source code in movinets_helper/utils.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
def get_chunks(l: List[Union[str, Path]], n: int) -> Iterable[Union[str, Path]]:
    r"""Yield successive n-sized chunks from l.

    Used to create n sublists from a list l.
    copied
     from: https://github.com/ferreirafabio/video2tfrecord/blob/7aa2c6312e2bc97baed7386b8c92c591769ee5bb/video2tfrecord.py#L55

    Args:
        l (List[Union[str, Path]]):
        n (int):

    Returns:
        Iterable[Union[str, Path]]:

    Example:

        >>> filenames_split = list(get_chunks(filenames, n_videos_in_record))

        To obtain the files from a dataframe.

        >>> next(get_chunks(dataset_df[["classes", "files"]].to_records(index=False), 10))
    """
    for i in range(0, len(l), n):
        yield l[i : i + n]

get_frame_count(path)

Counts the total number of frames in a video.

Parameters:

Name Type Description Default
path Path

description

 required

Returns:

Name Type Description
int int

description

Raises:

Type Description
AssertionError

if the path doesn't exist or it couldn't be loaded by some reason.
Example

get_video_capture_and_frame_count(sample_clip) 17

Source code in movinets_helper/utils.py 
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
def get_frame_count(path: Path) -> int:
    """Counts the total number of frames in a video.

    Args:
        path (Path): _description_

    Returns:
        int: _description_

    Raises:
        AssertionError: if the path doesn't exist or it couldn't be loaded
            by some reason.

    Example:

        >>> get_video_capture_and_frame_count(sample_clip)
        17
    """
    assert path.is_file(), "Couldn't find video file:" + path + ". Skipping video."

    cap = None

    if path:
        cap = cv2.VideoCapture(str(path))

    assert cap is not None, "Couldn't load video capture:" + path + ". Skipping video."

    # compute meta data of video
    if hasattr(cv2, "cv"):
        frame_count = int(cap.get(cv2.cv.CAP_PROP_FRAME_COUNT))
    else:
        frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))

    return frame_count

get_label_from_video_name(filename)

Extracts the label of the movement from the video filename.

Examples:

>>> path = PosixPath('.../chest-to-bar_6.mp4')
>>> get_label_from_video_name(path)
"chest-to-bar"
Source code in movinets_helper/utils.py 
89
90
91
92
93
94
95
96
97
98
def get_label_from_video_name(filename: Path) -> str:
    """Extracts the label of the movement from the video filename.

    Examples:

        >>> path = PosixPath('.../chest-to-bar_6.mp4')
        >>> get_label_from_video_name(path)
        "chest-to-bar"
    """
    return filename.stem.split("_")[0]

get_number_of_steps(samples, batch_size, epochs=1)

Obtain the number of steps.

Computes the number of steps per epoch and the total number of steps to be applied on the LearningScheduler (if used).

To be called both for train and validation/test.

Parameters:

Name Type Description Default
samples int

Number of examples in the dataset. If the data is splitted in train/test, each of them should be treated independently.

 required
batch_size int

Number of videos per step.

 required
epochs int

Number of epochs. If bigger than 1, the second returned value corresponds to the total number of steps the network will do. Defaults to 1.

 1

Returns:

Type Description
Tuple[int, int]

Tuple[int, int]: The first argument will either be the number of steps per epoch or the number of validation steps when calling fit on the model, and the second may be used to estimate the learning rate scheduler (when number of epochs > 1 and computing steps for the training samples).
Source code in movinets_helper/utils.py 
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
def get_number_of_steps(samples: int, batch_size: int, epochs: int = 1) -> Tuple[int, int]:
    """Obtain the number of steps.

    Computes the number of steps per epoch and the total number of steps
    to be applied on the LearningScheduler (if used).

    To be called both for train and validation/test.

    Args:
        samples (int): Number of examples in the dataset.
            If the data is splitted in train/test, each of them
            should be treated independently.
        batch_size (int): Number of videos per step.
        epochs (int, optional):
            Number of epochs. If bigger than 1, the second
            returned value corresponds to the total number of
            steps the network will do.
            Defaults to 1.

    Returns:
        Tuple[int, int]: The first argument will either be the number of steps
            per epoch or the number of validation steps when calling `fit` on
            the model, and the second may be used to estimate the learning
            rate scheduler (when number of epochs > 1 and computing steps
            for the training samples).
    """
    steps = samples // batch_size
    return steps, steps * epochs

get_top_k(probs, k=5, label_map=tf.constant(['']))

Outputs the top k model labels and probabilities on the given video.

Source code in movinets_helper/utils.py 
225
226
227
228
229
230
231
def get_top_k(probs, k=5, label_map=tf.constant([""])):
    """Outputs the top k model labels and probabilities on the given video."""
    top_predictions = tf.argsort(probs, axis=-1, direction='DESCENDING')[:k]
    top_labels = tf.gather(label_map, top_predictions, axis=-1)
    top_labels = [label.decode('utf8') for label in top_labels.numpy()]
    top_probs = tf.gather(probs, top_predictions, axis=-1).numpy()
    return tuple(zip(top_labels, top_probs))

load_model(path_to_model)

Load a model

Parameters:

Name Type Description Default
path_to_model Union[str, Path]

Path to the SavedModel directory.

 required

Returns:

Type Description
Functional

tf.keras.engine.functional.Functional: Model ready to predict.
Source code in movinets_helper/utils.py 
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
def load_model(path_to_model: Union[str, Path]) -> "tf.keras.engine.functional.Functional":
    r"""Load a model 

    Args:
        path_to_model (Union[str, Path]): Path to the SavedModel directory.

    Returns:
        tf.keras.engine.functional.Functional: Model ready to predict.
    """
    if isinstance(path_to_model, Path):
        path_to_model = str(path_to_model)

    keras_layer = hub.KerasLayer(path_to_model)

    inputs = tf.keras.layers.Input(
        shape=[None, None, None, 3],
        dtype=tf.float32
    )
    inputs = dict(image=inputs)
    outputs = keras_layer(inputs)

    model = tf.keras.Model(inputs, outputs)
    model.build([1, 1, 1, 1, 3])

    return model

load_video_tf(path)

Loads a video from an mp4 file and returns the decoded tensor.

Parameters:

Name Type Description Default
path str

Path to a video.

 required

Returns:

Type Description
tf.Tensor

tf.Tensor: Tensorflow's representation of the video.
Source code in movinets_helper/utils.py 
14
15
16
17
18
19
20
21
22
23
24
def load_video_tf(path: str) -> tf.Tensor:
    """Loads a video from an mp4 file and returns the decoded tensor.

    Args:
        path (str): Path to a video.

    Returns:
        tf.Tensor: Tensorflow's representation of the video.
    """
    video = tf.io.read_file(path)
    return tfio.experimental.ffmpeg.decode_video(video)

parse_video_from_bytes(video_encoded)

Parses a base64 encoded video to a tf.Tensor.

Used to read a video sent through a post request

Parameters:

Name Type Description Default
video_encoded _type_

base64 video encoded. This would be equivalent to:

with open(video_path, "rb") as video_file: text = base64.b64encode(video_file.read()) video_encoded = base64.b64decode(text)

 required

Returns:

Type Description
tf.Tensor

tf.Tensor: Video as tf.Tensor. Returns the same object that would be obtained from load_video_tf.
Source code in movinets_helper/utils.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def parse_video_from_bytes(video_encoded: bytes) -> tf.Tensor:
    """Parses a base64 encoded video to a tf.Tensor.

    Used to read a video sent through a post request

    Args:
        video_encoded (_type_): base64 video encoded.
            This would be equivalent to:

            >>> with open(video_path, "rb") as video_file:
            >>>    text = base64.b64encode(video_file.read())
            >>>    video_encoded = base64.b64decode(text)


    Returns:
        tf.Tensor: Video as tf.Tensor.
            Returns the same object that would be obtained from load_video_tf.
    """
    return tfio.experimental.ffmpeg.decode_video(base64.b64decode(video_encoded))

predict_top_k(model, video, k=5, label_map=tf.constant(['']))

Outputs the top k model labels and probabilities on the given video.

Parameters:

Name Type Description Default
model _type_

description

 required
video _type_

description

 required
k int

description. Defaults to 5.

 5
label_map _type_

description. Defaults to tf.constant([""]).

 tf.constant([''])

Returns:

Name Type Description
_type_

description
Source code in movinets_helper/utils.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def predict_top_k(model, video, k=5, label_map=tf.constant([""])):
    """Outputs the top k model labels and probabilities on the given video.

    Args:
        model (_type_): _description_
        video (_type_): _description_
        k (int, optional): _description_. Defaults to 5.
        label_map (_type_, optional): _description_. Defaults to tf.constant([""]).

    Returns:
        _type_: _description_
    """
    outputs = model.predict(video[tf.newaxis])[0]
    probs = tf.nn.softmax(outputs)
    return get_top_k(probs, k=k, label_map=label_map)

prepare_to_predict(video, resolution=224)

Adds a dimension to a video (and possibly resizes to the wanted resolution)

The model expects the videos with 5 dimensions. When a video was loaded with load_video_tf it will have only the 4 dimension expected, which would represent the batch size during training.

Parameters:

Name Type Description Default
video tf.Tensor

video as tf.Tensor, maybe loaded with load_video_tf.

 required
resolution int

If is different from None, resizes the video, to the expected resolution of the model. If set to None, doesn't do anything. Defaults to 224 (expected shape for a2 model).

 224

Returns:

Type Description
tf.Tensor

tf.Tensor: video ready to be passed to model.predict(video)
Source code in movinets_helper/utils.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def prepare_to_predict(video: tf.Tensor, resolution: int = 224) -> tf.Tensor:
    """Adds a dimension to a video (and possibly resizes to the wanted resolution)

    The model expects the videos with 5 dimensions. When a video
    was loaded with load_video_tf it will have only the 4 dimension
    expected, which would represent the batch size during training.

    Args:
        video (tf.Tensor): video as tf.Tensor, maybe loaded with load_video_tf.
        resolution (int): 
            If is different from None, resizes the video, to the expected
            resolution of the model. If set to None, doesn't do anything.
            Defaults to 224 (expected shape for a2 model).

    Returns:
        tf.Tensor: video ready to be passed to model.predict(video)
    """
    if resolution is not None:
        video = tf.image.resize(video, [resolution, resolution])

    return tf.expand_dims(video, 0)

split_train_test(dataset, train_size=0.8, seed=5678)

Simple function to split a dataset in train/test.

This functionality may be obtained from many other libraries, its just here for personal convinience.

Parameters:

Name Type Description Default
dataset pd.DataFrame

DataFrame with 3 columns: labels, files and classes.

 required
train_size float

Percentage of the sample for training, range [0, 1].

 0.8
seed int

Random seed to split the data. Defaults to 5678.

 5678

Returns:

Type Description
Tuple[pd.DataFrame, pd.DataFrame]

Tuple[pd.DataFrame, pd.DataFrame]: tran and test datasets.
Source code in movinets_helper/utils.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def split_train_test(
    dataset: pd.DataFrame, train_size: float = 0.8, seed: int = 5678
) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """Simple function to split a dataset in train/test.

    This functionality may be obtained from many other libraries,
    its just here for personal convinience.

    Args:
        dataset (pd.DataFrame):
            DataFrame with 3 columns: labels, files and classes.
        train_size (float):
            Percentage of the sample for training, range [0, 1].
        seed (int, optional): Random seed to split the data.
            Defaults to 5678.

    Returns:
        Tuple[pd.DataFrame, pd.DataFrame]: tran and test datasets.
    """
    train = dataset.sample(int(len(dataset) * train_size), random_state=seed)
    test = dataset.loc[set(dataset.index).difference(train.index), :]
    return train, test