Commit f83ac86a authored by Amir MOHAMMADI's avatar Amir MOHAMMADI
Browse files

port replay attack and replay mobile interfaces

parent 9a3f8488
Pipeline #48741 failed with stage
in 7 minutes and 22 seconds
......@@ -9,7 +9,6 @@ the link.
.. include:: links.rst
"""
from bob.pad.face.database import ReplayPadDatabase
from bob.pad.base.pipelines.vanilla_pad import DatabaseConnector
from bob.pad.face.database import ReplayAttackPadDatabase
database = DatabaseConnector(ReplayPadDatabase())
database = ReplayAttackPadDatabase()
from .database import VideoPadFile, VideoPadSample
from .replay import ReplayPadDatabase
from .replay_mobile import ReplayMobilePadDatabase
from .mifs import MIFSPadDatabase
from .database import VideoPadFile
from .database import VideoPadSample # noqa: F401
from .casiafasd import CasiaFasdPadDatabase
from .casiasurf import CasiaSurfPadDatabase
from .celeb_a import CELEBAPadDatabase
from .maskattack import MaskAttackPadDatabase
from .casiasurf import CasiaSurfPadDatabase
from .casiafasd import CasiaFasdPadDatabase
from .mifs import MIFSPadDatabase
from .replay_attack import ReplayAttackPadDatabase
from .replay_mobile import ReplayMobilePadDatabase
# gets sphinx autodoc done right - don't remove it
......@@ -26,7 +27,7 @@ def __appropriate__(*args):
__appropriate__(
VideoPadFile,
ReplayPadDatabase,
ReplayAttackPadDatabase,
ReplayMobilePadDatabase,
MIFSPadDatabase,
CELEBAPadDatabase,
......
from bob.pad.base.database import PadDatabase
from bob.pad.face.database import VideoPadFile
from bob.extension import rc
from bob.ip.facedetect import expected_eye_positions, BoundingBox
from bob.db.base.utils import convert_names_to_lowlevel
REPLAY_ATTACK_FRAME_SHAPE = (3, 240, 320)
class ReplayPadFile(VideoPadFile):
"""
A high level implementation of the File class for the REPLAY-ATTACK
database.
"""
def __init__(
self,
f,
**kwargs,
):
"""
Parameters
----------
f : object
An instance of the File class defined in the low level db interface
of the Replay database, in the bob.db.replay.models.py file.
"""
self.f = f
# this f is actually an instance of the File class that is defined in
# bob.db.replay.models and the PadFile class here needs client_id,
# path, attack_type, file_id for initialization. We have to convert
# information here and provide them to PadFile. attack_type is a little
# tricky to get here. Based on the documentation of PadFile: In cased
# of a spoofed data, this parameter should indicate what kind of
# spoofed attack it is. The default None value is interpreted that the
# PadFile is a genuine or real sample.
if f.is_real():
attack_type = None
else:
attack_type = "attack"
# attack_type is a string and I decided to make it like this for this
# particular database. You can do whatever you want for your own
# database.
super(ReplayPadFile, self).__init__(
client_id=f.client_id,
path=f.path,
attack_type=attack_type,
file_id=f.id,
**kwargs,
)
@property
def frame_shape(self):
"""Returns the size of each frame in this database.
Returns
-------
(int, int, int)
The (#Channels, Height, Width) which is (3, 240, 320).
"""
return REPLAY_ATTACK_FRAME_SHAPE
@property
def annotations(self):
"""
Return annotations as a dictionary of dictionaries.
If the file object has an attribute of annotation_directory, it will read
annotations from there instead of loading annotations that are shipped with the
database.
Returns
-------
annotations : :py:class:`dict`
A dictionary containing the annotations for each frame in the
video. Dictionary structure:
``annotations = {'1': frame1_dict, '2': frame1_dict, ...}``.Where
``frameN_dict = {'topleft': (row, col), 'bottomright': (row, col)}``
is the dictionary defining the coordinates of the face bounding box
in frame N.
"""
if self.annotation_directory is not None:
return super().annotations
# numpy array containing the face bounding box data for each video
# frame, returned data format described in the f.bbx() method of the
# low level interface
annots = self.f.bbx(directory=self.original_directory)
annotations = {} # dictionary to return
for fn, frame_annots in enumerate(annots):
topleft = (frame_annots[2], frame_annots[1])
bottomright = (
frame_annots[2] + frame_annots[4],
frame_annots[1] + frame_annots[3],
)
annotations[str(fn)] = {"topleft": topleft, "bottomright": bottomright}
size = (bottomright[0] - topleft[0], bottomright[1] - topleft[1])
bounding_box = BoundingBox(topleft, size)
annotations[str(fn)].update(expected_eye_positions(bounding_box))
return annotations
class ReplayPadDatabase(PadDatabase):
"""
A high level implementation of the Database class for the REPLAY-ATTACK
database.
"""
def __init__(
self,
# grandtest is the default protocol for this database
protocol="grandtest",
original_directory=rc.get("bob.db.replay.directory"),
original_extension=".mov",
annotation_directory=rc.get("bob.db.replay.annotation_dir"),
**kwargs,
):
"""
Parameters
----------
protocol : str or None
The name of the protocol that defines the default experimental
setup for this database.
original_directory : str
The directory where the original data of the database are stored.
original_extension : str
The file name extension of the original data.
kwargs
The arguments of the :py:class:`bob.bio.base.database.BioDatabase`
base class constructor.
"""
from bob.db.replay import Database as LowLevelDatabase
self.db = LowLevelDatabase()
# Since the high level API expects different group names than what the
# low level API offers, you need to convert them when necessary
self.low_level_group_names = (
"train",
"devel",
"test",
) # group names in the low-level database interface
self.high_level_group_names = (
"train",
"dev",
"eval",
) # names are expected to be like that in objects() function
# Always use super to call parent class methods.
super(ReplayPadDatabase, self).__init__(
name="replay",
protocol=protocol,
original_directory=original_directory,
original_extension=original_extension,
annotation_directory=annotation_directory,
**kwargs,
)
@property
def original_directory(self):
return self.db.original_directory
@original_directory.setter
def original_directory(self, value):
self.db.original_directory = value
def objects(
self, groups=None, protocol=None, purposes=None, model_ids=None, **kwargs
):
"""
This function returns lists of ReplayPadFile objects, which fulfill the
given restrictions.
Parameters
----------
groups : :obj:`str` or [:obj:`str`]
The groups of which the clients should be returned.
Usually, groups are one or more elements of
('train', 'dev', 'eval')
protocol : str
The protocol for which the clients should be retrieved.
The protocol is dependent on your database.
If you do not have protocols defined, just ignore this field.
purposes : :obj:`str` or [:obj:`str`]
The purposes for which File objects should be retrieved.
Usually it is either 'real' or 'attack'.
model_ids
This parameter is not supported in PAD databases yet
**kwargs
Returns
-------
files : [ReplayPadFile]
A list of ReplayPadFile objects.
"""
# Convert group names to low-level group names here.
groups = convert_names_to_lowlevel(
groups, self.low_level_group_names, self.high_level_group_names
)
# Since this database was designed for PAD experiments, nothing special
# needs to be done here.
files = self.db.objects(
protocol=protocol, groups=groups, cls=purposes, **kwargs
)
files = [
ReplayPadFile(
f,
original_directory=self.original_directory,
original_extension=self.original_extension,
annotation_directory=self.annotation_directory,
annotation_extension=self.annotation_extension,
annotation_type=self.annotation_type,
)
for f in files
]
return files
import logging
import numpy as np
from bob.extension import rc
from bob.extension.download import get_file
from bob.pad.base.database import FileListPadDatabase
from bob.pad.face.database import VideoPadSample
from bob.pipelines.transformers import Str_To_Types
from bob.pipelines.transformers import str_to_bool
from sklearn.pipeline import make_pipeline
logger = logging.getLogger(__name__)
def ReplayAttackPadDatabase(
protocol="grandtest",
selection_style=None,
max_number_of_frames=None,
step_size=None,
annotation_directory=None,
annotation_type=None,
fixed_positions=None,
**kwargs,
):
name = "pad-face-replay-attack-a8e31cc3.tar.gz"
dataset_protocols_path = get_file(
name,
[f"http://www.idiap.ch/software/bob/data/bob/bob.pad.face/{name}"],
cache_subdir="protocols",
file_hash="a8e31cc3",
)
if annotation_directory is None:
name = "annotations-replay-attack-mtcnn-3ecbfa3c.tar.xz"
annotation_directory = get_file(
name,
[f"http://www.idiap.ch/software/bob/data/bob/bob.pad.face/{name}"],
cache_subdir="annotations",
file_hash="3ecbfa3c",
)
annotation_type = "eyes-center"
transformer = VideoPadSample(
original_directory=rc.get("bob.db.replayattack.directory"),
annotation_directory=annotation_directory,
selection_style=selection_style,
max_number_of_frames=max_number_of_frames,
step_size=step_size,
)
database = FileListPadDatabase(
dataset_protocols_path,
protocol,
transformer=transformer,
**kwargs,
)
database.annotation_type = annotation_type
database.fixed_positions = fixed_positions
return database
import logging
import os
import numpy as np
from bob.extension import rc
from bob.extension.download import get_file
from bob.pad.base.database import FileListPadDatabase
from bob.pad.face.database import VideoPadSample
from bob.pipelines.transformers import Str_To_Types, str_to_bool
from bob.pipelines.transformers import Str_To_Types
from bob.pipelines.transformers import str_to_bool
from sklearn.pipeline import make_pipeline
from bob.extension.download import get_file
logger = logging.getLogger(__name__)
......@@ -35,17 +35,24 @@ def ReplayMobilePadDatabase(
fixed_positions=None,
**kwargs,
):
dataset_protocols_path = os.path.expanduser(
"~/temp/bob_data/datasets/pad-face-replay-mobile.tar.gz"
name = "pad-face-replay-mobile-586b7e81.tar.gz"
dataset_protocols_path = get_file(
name,
[f"http://www.idiap.ch/software/bob/data/bob/bob.pad.face/{name}"],
cache_subdir="protocols",
file_hash="586b7e81",
)
if annotation_directory is None:
name = "annotations-replaymobile-mtcnn-9cd6e452.tar.xz"
annotation_directory = get_file(
"replaymobile-mtcnn-annotations.tar.xz",
[
"http://www.idiap.ch/software/bob/data/bob/bob.db.replaymobile/master/replaymobile-mtcnn-annotations.tar.xz"
],
name,
[f"http://www.idiap.ch/software/bob/data/bob/bob.pad.face/{name}"],
cache_subdir="annotations",
file_hash="9cd6e452",
)
annotation_type = "eyes-center"
transformer = make_pipeline(
Str_To_Types(fieldtypes=dict(should_flip=str_to_bool)),
VideoPadSample(
......@@ -57,6 +64,7 @@ def ReplayMobilePadDatabase(
get_transform=get_rm_video_transform,
),
)
database = FileListPadDatabase(
dataset_protocols_path,
protocol,
......
......@@ -8,57 +8,43 @@ import bob.bio.base
from bob.bio.base.test.utils import db_available
@db_available("replay") # the name of the package
def test_replay():
# replay-attack is the name of the entry point
replay_database_instance = bob.bio.base.load_resource(
def test_replayattack():
database = bob.bio.base.load_resource(
"replay-attack",
"database",
preferred_package="bob.pad.face",
package_prefix="bob.pad.",
).database
try:
assert (
len(replay_database_instance.objects(groups=["train", "dev", "eval"]))
== 1200
)
assert len(replay_database_instance.objects(groups=["train", "dev"])) == 720
assert len(replay_database_instance.objects(groups=["train"])) == 360
assert (
len(
replay_database_instance.objects(
groups=["train", "dev", "eval"], protocol="grandtest"
)
)
== 1200
)
assert database.protocols() == ['digitalphoto', 'grandtest', 'highdef', 'mobile', 'photo', 'print', 'smalltest', 'video']
assert database.groups() == ["dev", "eval", "train"]
assert len(database.samples(groups=["train", "dev", "eval"])) == 1200
assert len(database.samples(groups=["train", "dev"])) == 720
assert len(database.samples(groups=["train"])) == 360
assert len(database.samples(groups=["train", "dev", "eval"])) == 1200
assert (
len(
replay_database_instance.objects(
groups=["train", "dev", "eval"],
protocol="grandtest",
purposes="real",
)
)
== 200
len(database.samples(groups=["train", "dev", "eval"], purposes="real")) == 200
)
assert (
len(
replay_database_instance.objects(
groups=["train", "dev", "eval"],
protocol="grandtest",
purposes="attack",
)
)
== 1000
len(database.samples(groups=["train", "dev", "eval"], purposes="attack")) == 1000
)
sample = database.sort(database.samples())[0]
try:
assert sample.annotations["0"] == {
"bottomright": [213, 219],
"topleft": [58, 108],
"leye": [118, 190],
"reye": [117, 137],
"mouthleft": [177, 144],
"mouthright": [180, 183],
"nose": [152, 164],
}
assert sample.data.shape == (20, 3, 240, 320)
assert sample.data[0][0, 0, 0] == 8
except IOError as e:
raise SkipTest(
"The database could not be queried; probably the db.sql3 file is missing. Here is the error: '%s'"
% e
)
raise SkipTest(e)
def test_replaymobile():
......@@ -94,6 +80,7 @@ def test_replaymobile():
"nose": [585, 358],
}
assert sample.data.shape == (20, 3, 720, 1280)
assert sample.data[0][0, 0, 0] == 13
except IOError as e:
raise SkipTest(e)
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment