replaymobile.py 10.8 KB
Newer Older
1
#!/usr/bin/env python
Yannick DAYER's avatar
Yannick DAYER committed
2
# Yannick Dayer <yannick.dayer@idiap.ch>
3

4
from bob.bio.base.database import CSVDataset, CSVToSampleLoaderBiometrics
5
from bob.pipelines.sample_loaders import AnnotationsLoader
Yannick DAYER's avatar
Yannick DAYER committed
6
from bob.pipelines.sample import DelayedSample
7
from bob.db.base.annotations import read_annotation_file
8
9
10
from bob.extension.download import get_file
from bob.io.video import reader
from bob.extension import rc
11

Yannick DAYER's avatar
Yannick DAYER committed
12
13
14
from sklearn.pipeline import make_pipeline
import functools
import os.path
Yannick DAYER's avatar
Yannick DAYER committed
15
import logging
Yannick DAYER's avatar
Yannick DAYER committed
16
import numpy
17

Yannick DAYER's avatar
Yannick DAYER committed
18
logger = logging.getLogger(__name__)
19

Yannick DAYER's avatar
[black]    
Yannick DAYER committed
20

21
def load_frame_from_file_replaymobile(file_name, frame, should_flip):
22
23
24
25
26
27
28
    """Loads a single frame from a video file for replay-mobile.

    This function uses bob's video reader utility that does not load the full
    video in memory to just access one frame.

    Parameters
    ----------
Yannick DAYER's avatar
Yannick DAYER committed
29

30
31
    file_name: str
        The video file to load the frames from
Yannick DAYER's avatar
Yannick DAYER committed
32

33
34
    frame: None or list of int
        The index of the frame to load.
35

36
37
38
    capturing device: str
        'mobile' devices' frames will be flipped vertically.
        Other devices' frames will not be flipped.
39

40
41
    Returns
    -------
42

43
44
    images: 3D numpy array
        The frame of the video in bob format (channel, height, width)
45
    """
Yannick DAYER's avatar
Yannick DAYER committed
46
    logger.debug(f"Reading frame {frame} from '{file_name}'")
47
48
49
50
51
    video_reader = reader(file_name)
    image = video_reader[frame]
    # Image captured by the 'mobile' device are flipped vertically.
    # (Images were captured horizontally and bob.io.video does not read the
    #   metadata correctly, whether it was on the right or left side)
52
    if should_flip:
53
        image = numpy.flip(image, 2)
Yannick DAYER's avatar
Yannick DAYER committed
54
    # Convert to bob format (channel, height, width)
55
56
    image = numpy.transpose(image, (0, 2, 1))
    return image
Yannick DAYER's avatar
Yannick DAYER committed
57

Yannick DAYER's avatar
[black]    
Yannick DAYER committed
58

59
60
class ReplayMobileCSVFrameSampleLoader(CSVToSampleLoaderBiometrics):
    """A loader transformer returning a specific frame of a video file.
61

62
    This is specifically tailored for replay-mobile. It uses a specific loader
Yannick DAYER's avatar
Yannick DAYER committed
63
    that processes the `should_flip` metadata to correctly orient the frames.
Yannick DAYER's avatar
Yannick DAYER committed
64
    """
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
65

Yannick DAYER's avatar
Yannick DAYER committed
66
67
    def __init__(
        self,
68
69
70
        dataset_original_directory="",
        extension="",
        reference_id_equal_subject_id=True,
Yannick DAYER's avatar
Yannick DAYER committed
71
72
    ):
        super().__init__(
73
74
75
76
77
            data_loader=None,
            extension=extension,
            dataset_original_directory=dataset_original_directory,
        )
        self.reference_id_equal_subject_id = reference_id_equal_subject_id
78
        self.references_list = []
79
80

    def convert_row_to_sample(self, row, header):
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
81
        """Creates a sample given a row of the CSV protocol definition."""
82
        fields = dict([[str(h).lower(), r] for h, r in zip(header, row)])
83
84

        if self.reference_id_equal_subject_id:
85
            fields["subject_id"] = fields["reference_id"]
86
        else:
87
            if "subject_id" not in fields:
88
                raise ValueError(f"`subject_id` not available in {header}")
89
        if "should_flip" not in fields:
90
            raise ValueError(f"`should_flip` not available in {header}")
91
92
93
        if "purpose" not in fields:
            raise ValueError(f"`purpose` not available in {header}")

Yannick DAYER's avatar
[black]    
Yannick DAYER committed
94
        kwargs = {k: fields[k] for k in fields.keys() - {"id", "should_flip"}}
95
96

        # Retrieve the references list
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
97
98
99
100
        if (
            fields["purpose"].lower() == "enroll"
            and fields["reference_id"] not in self.references_list
        ):
101
102
103
104
            self.references_list.append(fields["reference_id"])
        # Set the references list in the probes for vanilla-biometrics
        if fields["purpose"].lower() != "enroll":
            if fields["attack_type"]:
Yannick DAYER's avatar
Yannick DAYER committed
105
106
                # Attacks are only compare to their target (no `spoof_neg`)
                kwargs["references"] = [fields["reference_id"]]
107
108
            else:
                kwargs["references"] = self.references_list
109
        # One row leads to multiple samples (different frames)
Yannick DAYER's avatar
Yannick DAYER committed
110
        return DelayedSample(
111
112
            functools.partial(
                load_frame_from_file_replaymobile,
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
113
114
115
                file_name=os.path.join(
                    self.dataset_original_directory, fields["path"] + self.extension
                ),
Yannick DAYER's avatar
Yannick DAYER committed
116
                frame=int(fields["frame"]),
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
117
                should_flip=fields["should_flip"] == "TRUE",
Yannick DAYER's avatar
Yannick DAYER committed
118
            ),
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
119
            key=fields["id"],
120
            **kwargs,
Yannick DAYER's avatar
Yannick DAYER committed
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
def read_frame_annotation_file_replaymobile(file_name, frame, annotations_type="json"):
    """Returns the bounding-box for one frame of a video file of replay-mobile.

    Given an annnotation file location and a frame number, returns the bounding
    box coordinates corresponding to the frame.

    The replay-mobile annotation files are composed of 4 columns and N rows for
    N frames of the video:

    120 230 40 40
    125 230 40 40
    ...
    <x> <y> <w> <h>

    Parameters
    ----------

    file_name: str
        The annotation file name (relative to annotations_path).

    frame: int
        The video frame index.
    """
    logger.debug(f"Reading annotation file '{file_name}', frame {frame}.")

Yannick DAYER's avatar
[black]    
Yannick DAYER committed
149
150
151
    video_annotations = read_annotation_file(
        file_name, annotation_type=annotations_type
    )
152
    # read_annotation_file returns an ordered dict with str keys as frame number
153
    frame_annotations = video_annotations[str(frame)]
154
    if frame_annotations is None:
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
155
156
157
        logger.warning(
            f"Annotation for file '{file_name}' at frame {frame} was 'null'."
        )
158
    return frame_annotations
159

Yannick DAYER's avatar
[black]    
Yannick DAYER committed
160

161
162
163
164
165
166
167
168
class FrameBoundingBoxAnnotationLoader(AnnotationsLoader):
    """A transformer that adds bounding-box to a sample from annotations files.

    Parameters
    ----------

    annotation_directory: str or None
    """
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
169
170
171

    def __init__(
        self, annotation_directory=None, annotation_extension=".json", **kwargs
172
173
174
175
    ):
        super().__init__(
            annotation_directory=annotation_directory,
            annotation_extension=annotation_extension,
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
176
            **kwargs,
Yannick DAYER's avatar
Yannick DAYER committed
177
        )
178

179
    def transform(self, X):
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
180
        """Adds the bounding-box annotations to a series of samples."""
181
182
        if self.annotation_directory is None:
            return None
183

184
185
        annotated_samples = []
        for x in X:
186
            # Adds the annotations as delayed_attributes, loading them when needed
187
188
189
190
191
192
193
            annotated_samples.append(
                DelayedSample(
                    x._load,
                    parent=x,
                    delayed_attributes=dict(
                        annotations=functools.partial(
                            read_frame_annotation_file_replaymobile,
194
                            file_name=f"{self.annotation_directory}:{x.path}{self.annotation_extension}",
195
                            frame=int(x.frame),
196
                            annotations_type=self.annotation_type,
197
198
199
200
                        )
                    ),
                )
            )
201

202
        return annotated_samples
Amir MOHAMMADI's avatar
Amir MOHAMMADI committed
203

204

205
206
class ReplayMobileBioDatabase(CSVDataset):
    """Database interface that loads a csv definition for replay-mobile
Yannick DAYER's avatar
Yannick DAYER committed
207

208
209
210
211
    Looks for the protocol definition files (structure of CSV files). If not
    present, downloads them.
    Then sets the data and annotation paths from __init__ parameters or from
    the configuration (``bob config`` command).
Yannick DAYER's avatar
Yannick DAYER committed
212

213
214
    Parameters
    ----------
Yannick DAYER's avatar
Yannick DAYER committed
215

216
    protocol_name: str
217
        The protocol to use. Must be a sub-folder of ``protocol_definition_path``
Yannick DAYER's avatar
Yannick DAYER committed
218

219
    protocol_definition_path: str or None
220
        Specifies a path where to fetch the database definition from.
Yannick DAYER's avatar
Yannick DAYER committed
221
        (See :py:func:`bob.extension.download.get_file`)
222
223
        If None: Downloads the file in the path from ``bob_data_folder`` config.
        If None and the config does not exist: Downloads the file in ``~/bob_data``.
Yannick DAYER's avatar
Yannick DAYER committed
224

225
226
227
228
229
230
    data_path: str or None
        Overrides the config-defined data location.
        If None: uses the ``bob.db.replaymobile.directory`` config.
        If None and the config does not exist, set as cwd.

    annotation_path: str or None
231
232
233
234
        Specifies a path where the annotation files are located.
        If None: Downloads the files to the path poited by the
        ``bob.db.replaymobile.annotation_directory`` config.
        If None and the config does not exist: Downloads the file in ``~/bob_data``.
235
    """
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
236

237
238
    def __init__(
        self,
239
        protocol="grandtest",
240
241
        protocol_definition_path=None,
        data_path=None,
242
243
244
        data_extension=".mov",
        annotations_path=None,
        annotations_extension=".json",
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
245
        **kwargs,
246
247
248
    ):
        if protocol_definition_path is None:
            # Downloading database description files if it is not specified
249
250
251
252
            proto_def_name = "bio-face-replaymobile-img-3a584a97.tar.gz"
            proto_def_urls = [
                f"https://www.idiap.ch/software/bob/data/bob/bob.bio.face/{proto_def_name}",
                f"http://www.idiap.ch/software/bob/data/bob/bob.bio.face/{proto_def_name}",
253
            ]
254
            protocol_definition_path = get_file(
255
256
257
258
                filename=proto_def_name,
                urls=proto_def_urls,
                cache_subdir="datasets",
                file_hash="3a584a97",
259
            )
260
261
262

        if data_path is None:
            data_path = rc.get("bob.db.replaymobile.directory", "")
263
264
265
266
267
268
        if data_path == "":
            logger.warning(
                "Raw data path is not configured. Please set "
                "'bob.db.replaymobile.directory' with the 'bob config set' command. "
                "Will now attempt with current directory."
            )
269

270
        if annotations_path is None:
271
272
273
274
            annot_name = "annotations-replaymobile-mtcnn-9cd6e452.tar.xz"
            annot_urls = [
                f"https://www.idiap.ch/software/bob/data/bob/bob.pad.face/{annot_name}",
                f"http://www.idiap.ch/software/bob/data/bob/bob.pad.face/{annot_name}",
275
            ]
Yannick DAYER's avatar
Yannick DAYER committed
276
            annotations_path = get_file(
277
278
                filename=annot_name,
                urls=annot_urls,
Yannick DAYER's avatar
Yannick DAYER committed
279
280
                cache_subdir="annotations",
                file_hash="9cd6e452",
281
282
            )

Yannick DAYER's avatar
[black]    
Yannick DAYER committed
283
284
285
        logger.info(
            f"Database: Will read CSV protocol definitions in '{protocol_definition_path}'."
        )
286
287
        logger.info(f"Database: Will read raw data files in '{data_path}'.")
        logger.info(f"Database: Will read annotation files in '{annotations_path}'.")
288
        super().__init__(
289
290
291
            name="replaymobile-img",
            protocol=protocol,
            dataset_protocol_path=protocol_definition_path,
292
293
294
            csv_to_sample_loader=make_pipeline(
                ReplayMobileCSVFrameSampleLoader(
                    dataset_original_directory=data_path,
295
                    extension=data_extension,
296
297
                ),
                FrameBoundingBoxAnnotationLoader(
298
299
                    annotation_directory=annotations_path,
                    annotation_extension=annotations_extension,
300
301
                ),
            ),
302
            fetch_probes=False,
Yannick DAYER's avatar
[black]    
Yannick DAYER committed
303
            **kwargs,
304
        )
305
        self.annotation_type = "eyes-center"
306
        self.fixed_positions = None