Source code for

#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
# @author: Manuel Guenther <>
# @author: Pavel Korshunov <>
# @date: Wed 19 Aug 13:43:21 2015

import as bio_load
import bob.measure
import numpy
import os
import sys
import tarfile

import logging

logger = logging.getLogger("bob.pad.base")

from .FileSelector import FileSelector

from import utils

def _compute_scores(algorithm, extractor, toscore_objects, allow_missing_files):
    """Compute scores for the given list of objects using provided algorithm.
    # the scores to be computed
    scores = []

    # Loops over the toscore sets
    for i, toscore_element in enumerate(toscore_objects):
        # filter missing files
        if allow_missing_files and not os.path.exists(toscore_element):
            # we keep NaN score for such elements
            scores.insert(i, [numpy.nan])
        # read toscore
        if algorithm.performs_projection:
            toscore = algorithm.read_feature(toscore_element)
            toscore = extractor.read_feature(toscore_element)
        # compute score
        if isinstance(toscore, list) or isinstance(toscore[0], numpy.ndarray):
            scores.insert(i, algorithm.score_for_multiple_projections(toscore))
            scores.insert(i, algorithm.score(toscore))
    # Returns the scores
    return scores

def _open_to_read(score_file):
    """Checks for the existence of the normal and the compressed version of the file,
    and calls :py:func:`` for the existing one."""
    if not os.path.exists(score_file):
        score_file += '.tar.bz2'
        if not os.path.exists(score_file):
            raise IOError("The score file '%s' cannot be found. Aborting!" % score_file)
    return bio_load.open_file(score_file)

def _open_to_write(score_file, write_compressed):
    """Opens the given score file for writing. If write_compressed is set to ``True``, a file-like structure is returned."""
    if write_compressed:
        if sys.version_info[0] <= 2:
            import StringIO
            f = StringIO.StringIO()
            import io
            f = io.BytesIO()
        score_file += '.tar.bz2'
        f = open(score_file, 'w')

    return f

def _write(f, data, write_compressed):
    """Writes the given data to file, after converting it to the required type."""
    if write_compressed:
        if sys.version_info[0] > 2:
            data = str.encode(data)


def _close_written(score_file, f, write_compressed):
    """Closes the file f that was opened with :py:func:`_open_to_read`"""
    if write_compressed:
        tarinfo = tarfile.TarInfo(os.path.basename(score_file))
        tarinfo.size = len(f.buf if sys.version_info[0] <= 2 else f.getbuffer())
        tar =, 'w')
        tar.addfile(tarinfo, f)
    # close the file

def _save_scores(score_file, scores, toscore_objects, write_compressed=False):
    """Saves the scores of one model into a text file that can be interpreted by :py:func:`bob.measure.load.split_four_column`."""
    assert len(toscore_objects) == len(scores)

    # open file for writing
    f = _open_to_write(score_file, write_compressed)

    # write scores in four-column format as string
    for i, toscore_object in enumerate(toscore_objects):
        id_str = (str(toscore_object.client_id)).zfill(3)
        sample_name = str(toscore_object.make_path())

        # scores[i] is a list, so
        # each sample is allowed to have multiple scores
        for score in scores[i]:
            if not toscore_object.attack_type or toscore_object.attack_type == "None":
                _write(f, "%s %s %s %.12f\n" % (id_str, id_str, sample_name, score), write_compressed)
                attackname = toscore_object.attack_type
                _write(f, "%s %s %s %.12f\n" % (id_str, attackname, sample_name, score), write_compressed)

    _close_written(score_file, f, write_compressed)

def _scores_all(algorithm, extractor, group, force, allow_missing_files=False, write_compressed=False):
    """Computes scores for all (real, attack) files in a given group using the provided algorithm."""
    # the file selector object
    fs = FileSelector.instance()"- Scoring: computing scores for group '%s'", group)

    # both real and attack objects
    current_toscore_objects = fs.toscore_objects(group)
    type_objects = ['real', 'attack']

    total_scores = []
    one_score_file_exists = False
    for i in range(0, 2):
        current_objects = current_toscore_objects[i]
        obj_type = type_objects[i]"- Scoring: len of current_objects: %d", len(current_objects))"- Scoring: type of current_objects: %s", obj_type)
        # test if the file is already there
        score_file = fs.score_file_for_type(group, obj_type)
        if utils.check_file(score_file, force):
            logger.warn("Score file '%s' already exists.", score_file)
            total_scores = []
            one_score_file_exists = True
            # get the attack files
            current_files = fs.get_paths(current_objects, 'projected' if algorithm.performs_projection else 'extracted')
            # compute scores for the list of File objects
            cur_scores = _compute_scores(algorithm, extractor, current_files, allow_missing_files)
            total_scores += cur_scores
            # Save scores to text file
            _save_scores(score_file, cur_scores, current_objects, write_compressed)

    if total_scores != [] and not utils.check_file(fs.score_file_combined(group), force):
        # save all scores together in one file
        if one_score_file_exists:
            logger.warn("Since at least one score file already pre-existed, "
                        "we skip combining individual score files together. "
                        "You can do it manually, using 'cat' or similar utilities.")
            _save_scores(fs.score_file_combined(group), total_scores,
                         current_toscore_objects[0]+current_toscore_objects[1], write_compressed)

[docs]def compute_scores(algorithm, extractor, force=False, groups=['dev', 'eval'], allow_missing_files=False, write_compressed=False): """Computes the scores for the given groups. This function computes all scores for the experiment and writes them to score files. By default, scores are computed for both groups ``'dev'`` and ``'eval'``. **Parameters:** algorithm : py:class:`` or derived The algorithm, used for enrolling model and writing them to file. extractor : py:class:`` or derived force : bool If given, files are regenerated, even if they already exist. groups : some of ``('dev', 'eval')`` The list of groups, for which scores should be computed. write_compressed : bool If enabled, score files are compressed as ``.tar.bz2`` files. """ # the file selector object fs = FileSelector.instance() # load the projector if needed if algorithm.requires_projector_training: algorithm.load_projector(fs.projector_file) for group in groups: _scores_all(algorithm, extractor, group, force, allow_missing_files, write_compressed)