Coverage for src/bob/bio/face/database/replaymobile.py: 86%
85 statements
« prev ^ index » next coverage.py v7.6.0, created at 2024-07-13 00:04 +0200
1#!/usr/bin/env python
2# Yannick Dayer <yannick.dayer@idiap.ch>
4import functools
5import logging
6import os.path
8from typing import Optional
10import imageio
11import numpy
13from clapper.rc import UserDefaults
14from sklearn.base import BaseEstimator
15from sklearn.pipeline import make_pipeline
17import bob.io.image
19from bob.bio.base.database import CSVDatabase, FileSampleLoader
20from bob.bio.base.database.utils import download_file
21from bob.bio.base.utils.annotations import read_annotation_file
22from bob.pipelines import hash_string
23from bob.pipelines.sample import DelayedSample
25logger = logging.getLogger(__name__)
26rc = UserDefaults("bobrc.toml")
28read_annotation_file = functools.lru_cache()(read_annotation_file)
31def load_frame_from_file_replaymobile(file_name, frame, should_flip):
32 """Loads a single frame from a video file for replay-mobile.
34 This function uses bob's video reader utility that does not load the full
35 video in memory to just access one frame.
37 Parameters
38 ----------
40 file_name: str
41 The video file to load the frames from
43 frame: None or list of int
44 The index of the frame to load.
46 capturing device: str
47 ``mobile`` devices' frames will be flipped vertically.
48 Other devices' frames will not be flipped.
50 Returns
51 -------
53 images: 3D numpy array
54 The frame of the video in bob format (channel, height, width)
55 """
56 logger.debug(f"Reading frame {frame} from '{file_name}'")
58 video_reader = imageio.get_reader(file_name)
59 image = video_reader.get_data(frame)
60 # Convert to bob format (channel, height, width)
61 image = bob.io.image.to_bob(image)
63 # Image captured by the 'mobile' device are flipped vertically.
64 # (Images were captured horizontally and bob.io.video does not read the
65 # metadata correctly, whether it was on the right or left side)
66 if not should_flip:
67 # after changing from bob.io.video to imageio-ffmpeg, the tablet
68 # videos should be flipped to match previous behavior.
69 image = numpy.flip(image, 2)
71 return image
74class ReplayMobileCSVFrameSampleLoader(FileSampleLoader):
75 """A loader transformer returning a specific frame of a video file.
77 This is specifically tailored for replay-mobile. It uses a specific loader
78 that processes the `should_flip` metadata to correctly orient the frames.
79 """
81 def __init__(
82 self,
83 dataset_original_directory="",
84 extension="",
85 template_id_equal_subject_id=True,
86 ):
87 super().__init__(
88 data_loader=lambda: None,
89 extension=extension,
90 dataset_original_directory=dataset_original_directory,
91 )
92 self.template_id_equal_subject_id = template_id_equal_subject_id
94 def transform(self, samples):
95 """Creates a sample given a row of the CSV protocol definition."""
96 output = []
98 for sample in samples:
99 if not self.template_id_equal_subject_id and not hasattr(
100 sample, "subject_id"
101 ):
102 raise ValueError(f"`subject_id` not available in {sample}")
103 if not hasattr(sample, "should_flip"):
104 raise ValueError(f"`should_flip` not available in {sample}")
106 subject_id = (
107 sample.subject_id
108 if not self.template_id_equal_subject_id
109 or not hasattr(sample, "template_id") # e.g. in train set
110 else sample.template_id
111 )
113 # One row creates one samples (=> one comparison because of `is_sparse`)
114 should_flip = sample.should_flip.lower() == "true"
115 new_s = DelayedSample(
116 functools.partial(
117 load_frame_from_file_replaymobile,
118 file_name=os.path.join(
119 self.dataset_original_directory,
120 sample.path + self.extension,
121 ),
122 frame=int(sample.frame),
123 should_flip=should_flip,
124 ),
125 key=sample.key,
126 should_flip=should_flip,
127 subject_id=subject_id,
128 parent=sample,
129 )
130 output.append(new_s)
131 return output
134def read_frame_annotation_file_replaymobile(
135 file_name, frame, annotations_type="json"
136):
137 """Returns the bounding-box for one frame of a video file of replay-mobile.
139 Given an annotation file location and a frame number, returns the bounding
140 box coordinates corresponding to the frame.
142 The replay-mobile annotation files are composed of 4 columns and N rows for
143 N frames of the video:
145 120 230 40 40
146 125 230 40 40
147 ...
148 <x> <y> <w> <h>
150 Parameters
151 ----------
153 file_name: str
154 The annotation file name (relative to annotations_path).
156 frame: int
157 The video frame index.
158 """
159 logger.debug(f"Reading annotation file '{file_name}', frame {frame}.")
161 video_annotations = read_annotation_file(
162 file_name, annotation_type=annotations_type
163 )
164 # read_annotation_file returns an ordered dict with str keys as frame number
165 frame_annotations = video_annotations[str(frame)]
166 if frame_annotations is None:
167 logger.warning(
168 f"Annotation for file '{file_name}' at frame {frame} was 'null'."
169 )
170 return frame_annotations
173class FrameBoundingBoxAnnotationLoader(BaseEstimator):
174 """A transformer that adds bounding-box to a sample from annotations files.
176 Parameters
177 ----------
179 annotation_directory: str or None
180 """
182 def __init__(
183 self,
184 annotation_directory: Optional[str] = None,
185 annotation_extension: str = ".json",
186 **kwargs,
187 ):
188 self.annotation_directory = annotation_directory
189 self.annotation_extension = annotation_extension
190 self.annotation_type = annotation_extension.replace(".", "")
192 def transform(self, X):
193 """Adds the bounding-box annotations to a series of samples."""
194 if self.annotation_directory is None:
195 return None
197 annotated_samples = []
198 for x in X:
199 # Adds the annotations as delayed_attributes, loading them when needed
200 annotated_samples.append(
201 DelayedSample.from_sample(
202 x,
203 delayed_attributes=dict(
204 annotations=functools.partial(
205 read_frame_annotation_file_replaymobile,
206 file_name=f"{self.annotation_directory}:{x.path}{self.annotation_extension}",
207 frame=int(x.frame),
208 annotations_type=self.annotation_type,
209 )
210 ),
211 )
212 )
214 return annotated_samples
216 def _more_tags(self):
217 return {
218 "requires_fit": False,
219 }
222class ReplayMobileBioDatabase(CSVDatabase):
223 """Database interface that loads a csv definition for replay-mobile
225 Looks for the protocol definition files (structure of CSV files). If not
226 present, downloads them.
227 Then sets the data and annotation paths from __init__ parameters or from
228 the configuration (``bob config`` command).
230 Parameters
231 ----------
233 protocol_name: str
234 The protocol to use. Must be a sub-folder of ``protocol_definition_path``
236 protocol_definition_path: str or None
237 Specifies a path where to fetch the database definition from.
238 If None: Downloads the file in the path from ``bob_data_folder`` config.
239 If None and the config does not exist: Downloads the file in ``~/bob_data``.
241 data_path: str or None
242 Overrides the config-defined data location.
243 If None: uses the ``bob.db.replaymobile.directory`` config.
244 If None and the config does not exist, set as cwd.
246 annotation_path: str or None
247 Specifies a path where the annotation files are located.
248 If None: Downloads the files to the path pointed by the
249 ``bob.db.replaymobile.annotation_directory`` config.
250 If None and the config does not exist: Downloads the file in ``~/bob_data``.
251 """
253 name = "replaymobile"
254 category = "face"
255 dataset_protocols_name = "replaymobile.tar.gz"
256 dataset_protocols_urls = [
257 "https://www.idiap.ch/software/bob/databases/latest/face/replaymobile-354f3301.tar.gz",
258 "http://www.idiap.ch/software/bob/databases/latest/face/replaymobile-354f3301.tar.gz",
259 ]
260 dataset_protocols_hash = "354f3301"
262 def __init__(
263 self,
264 protocol="grandtest",
265 protocol_definition_path=None,
266 data_path=rc.get("bob.db.replaymobile.directory", ""),
267 data_extension=rc.get("bob.db.replaymobile.extension", ".mov"),
268 annotations_path=None,
269 annotations_extension=".json",
270 **kwargs,
271 ):
272 if data_path == "":
273 logger.warning(
274 "Raw data path is not configured. Please set "
275 "'bob.db.replaymobile.directory' with the 'bob config set' command. "
276 "Will now attempt with current directory."
277 )
279 if annotations_path is None:
280 annot_hash = "9cd6e452"
281 annot_name = f"annotations-replaymobile-mtcnn-{annot_hash}.tar.xz"
282 annot_urls = [
283 f"https://www.idiap.ch/software/bob/data/bob/bob.pad.face/{annot_name}",
284 f"http://www.idiap.ch/software/bob/data/bob/bob.pad.face/{annot_name}",
285 ]
286 annotations_path = download_file(
287 urls=annot_urls,
288 destination_sub_directory="annotations",
289 destination_filename=annot_name,
290 checksum=annot_hash,
291 extract=True,
292 )
293 annotations_path = os.path.join(
294 annotations_path,
295 "replaymobile-mtcnn-annotations",
296 )
298 logger.info(
299 f"Database: Will read CSV protocol definitions in '{protocol_definition_path}'."
300 )
301 logger.info(f"Database: Will read raw data files in '{data_path}'.")
302 logger.info(
303 f"Database: Will read annotation files in '{annotations_path}'."
304 )
305 super().__init__(
306 name="replaymobile",
307 protocol=protocol,
308 dataset_protocols_path=protocol_definition_path,
309 transformer=make_pipeline(
310 ReplayMobileCSVFrameSampleLoader(
311 dataset_original_directory=data_path,
312 extension=data_extension,
313 ),
314 FrameBoundingBoxAnnotationLoader(
315 annotation_directory=annotations_path,
316 annotation_extension=annotations_extension,
317 ),
318 ),
319 score_all_vs_all=False,
320 **kwargs,
321 )
322 self.annotation_type = "eyes-center"
323 self.fixed_positions = None
324 self.hash_fn = hash_string