#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
import os
import warnings
import logging
from . import utils
from .file import File
from .utils import check_parameters_for_validity, \
check_parameter_for_validity, \
convert_names_to_highlevel, \
convert_names_to_lowlevel, \
file_names, \
sort_files
logger = logging.getLogger(__name__)
class FileDatabase(object):
"""Low-level File-based Database API to be used within Bob.
Not all Databases in Bob need to inherit from this class. Use this class
only if in your database one sample correlates to one actual file.
Attributes
----------
original_directory : str
The directory where the raw files are located.
original_extension : str
The extension of raw data files, e.g. ``.png``.
"""
def __init__(self, original_directory, original_extension, **kwargs):
super(FileDatabase, self).__init__(**kwargs)
self.original_directory = original_directory
self.original_extension = original_extension
[docs] def original_file_names(self, files):
"""Returns the full path of the original data of the given File objects.
Parameters
----------
files : list of :py:class:`bob.db.base.File`
The list of file object to retrieve the original data file names for.
Returns
-------
list of :obj:`str`
The paths extracted for the files, in the same order.
"""
if self.original_directory is None:
logger.warning(
'self.original_directory was not provided (must not be None)!')
if self.original_extension is None:
logger.warning(
'self.original_extension was not provided (must not be None)!')
return file_names(
files, self.original_directory, self.original_extension)
[docs] def original_file_name(self, file):
"""This function returns the original file name for the given File
object.
Parameters
----------
file
:py:class:`bob.db.base.File` or a derivative
The File objects for which the file name should be retrieved
Returns
-------
str
The original file name for the given :py:class:`bob.db.base.File`
object.
Raises
------
ValueError
if the file is not found.
"""
# check if directory is set
if not self.original_directory or not self.original_extension:
logger.warning(
"The original_directory and/or the original_extension were not"
" specified in the constructor.")
# extract file name
file_name = file.make_path(
self.original_directory, self.original_extension)
if not self.check_existence or os.path.exists(file_name):
return file_name
raise ValueError("The file '%s' was not found. Please check the "
"original directory '%s' and extension '%s'?" % (
file_name,
self.original_directory,
self.original_extension))
# Deprecated Methods below
def check_parameters_for_validity(self, parameters, parameter_description,
valid_parameters,
default_parameters=None):
warnings.warn("check_parameters_for_validity is deprecated. Please use "
"the equivalent function in bob.db.base.utils",
DeprecationWarning, stacklevel=2)
return check_parameters_for_validity(parameters, parameter_description,
valid_parameters,
default_parameters)
def check_parameter_for_validity(self, parameter, parameter_description,
valid_parameters, default_parameter=None):
warnings.warn("check_parameter_for_validity is deprecated. Please use the "
"equivalent function in bob.db.base.utils",
DeprecationWarning, stacklevel=2)
return check_parameter_for_validity(parameter, parameter_description,
valid_parameters,
default_parameter)
def convert_names_to_highlevel(self, names, low_level_names,
high_level_names):
warnings.warn("convert_names_to_highlevel is deprecated. Please use the "
"equivalent function in bob.db.base.utils",
DeprecationWarning, stacklevel=2)
return convert_names_to_highlevel(names, low_level_names,
high_level_names)
def convert_names_to_lowlevel(self, names, low_level_names,
high_level_names):
warnings.warn("convert_names_to_lowlevel is deprecated. Please use the "
"equivalent function in bob.db.base.utils",
DeprecationWarning, stacklevel=2)
return convert_names_to_lowlevel(names, low_level_names,
high_level_names)
def file_names(self, files, directory, extension):
warnings.warn("file_names is deprecated. Please use the "
"equivalent function in bob.db.base.utils",
DeprecationWarning, stacklevel=2)
return file_names(files, directory, extension)
def sort(self, files):
warnings.warn("sort is deprecated. Please use "
"sort_files in bob.db.base.utils",
DeprecationWarning, stacklevel=2)
return sort_files(files)
class Database(FileDatabase):
"""This class is deprecated. New databases should use the
:py:class:`bob.db.base.FileDatabase` class if required"""
def __init__(self, original_directory=None, original_extension=None,
**kwargs):
warnings.warn("The bob.db.base.Database class is deprecated. "
"Please use bob.db.base.FileDatabase instead.",
DeprecationWarning, stacklevel=2)
super(Database, self).__init__(original_directory, original_extension,
**kwargs)
class SQLiteBaseDatabase(object):
"""This class can be used for handling SQL databases.
It opens an SQL database in a read-only mode and keeps it opened during the
whole session.
Parameters
----------
sqlite_file : str
The file name (including full path) of the SQLite file to read or
generate.
file_class : :py:class:`bob.db.base.File`
The ``File`` class, which needs to be derived from
:py:class:`bob.db.base.File`. This is required to be able to
:py:meth:`query` the databases later on.
Attributes
----------
m_file_class : :py:class:`bob.db.base.File`
The `file_class` parameter is kept in this attribute.
m_session : object
The SQL session object.
m_sqlite_file : str
The `sqlite_file` parameter is kept in this attribute.
"""
def __init__(self, sqlite_file, file_class, **kwargs):
super(SQLiteBaseDatabase, self).__init__(**kwargs)
self.m_sqlite_file = sqlite_file
if not os.path.exists(sqlite_file):
self.m_session = None
else:
self.m_session = utils.session_try_readonly('sqlite', sqlite_file)
# assert the given file class is derived from the File class
assert issubclass(file_class, File)
self.m_file_class = file_class
def __getstate__(self):
state = self.__dict__.copy()
state["m_session"] = None
return state
def __setstate__(self, state):
self.__dict__.update(state)
if os.path.exists(self.m_sqlite_file):
self.m_session = utils.session_try_readonly('sqlite', self.m_sqlite_file)
def __del__(self):
"""Closes the connection to the database."""
if self.is_valid():
# do some magic to close the connection to the database file
try:
# Since the dispose function re-creates a pool
# which might fail in some conditions, e.g., when this
# destructor is called during the exit of the python
# interpreter
self.m_session.close()
self.m_session.bind.dispose()
except (TypeError, AttributeError, KeyError):
# ... I can just ignore the according exception...
pass
[docs] def is_valid(self):
"""Returns if a valid session has been opened for reading the database.
"""
return self.m_session is not None
[docs] def assert_validity(self):
"""Raise a RuntimeError if the database back-end is not available."""
if not self.is_valid():
raise IOError(
"Database of type 'sqlite' cannot be found at expected "
"location '%s'." % self.m_sqlite_file)
[docs] def query(self, *args):
"""Creates a query to the database using the given arguments."""
self.assert_validity()
return self.m_session.query(*args)
[docs] def files(self, ids, preserve_order=True):
"""Returns a list of ``File`` objects with the given file ids
Parameters
----------
ids : :obj:`list` or :obj:`tuple`
The ids of the object in the database table "file". This object
should be a python iterable (such as a tuple or list).
preserve_order : bool
If True (the default) the order of elements is preserved, but the
execution time increases.
Returns
-------
list
a list (that may be empty) of ``File`` objects.
"""
file_objects = self.query(self.m_file_class).filter(
self.m_file_class.id.in_(ids))
if not preserve_order:
return list(file_objects)
else:
path_dict = {}
for f in file_objects:
path_dict[f.id] = f
return [path_dict[id] for id in ids]
[docs] def paths(self, ids, prefix=None, suffix=None, preserve_order=True):
"""Returns a full file paths considering particular file ids
Parameters
----------
ids : :obj:`list` or :obj`tuple`
The ids of the object in the database table "file". This object should
be a python iterable (such as a tuple or list).
prefix : :obj:`str`, optional
The bit of path to be prepended to the filename stem
suffix : :obj:`str`, optional
The extension determines the suffix that will be appended to the
filename stem.
preserve_order : bool
If True (the default) the order of elements is preserved, but the
execution time increases.
Returns
-------
list
A list (that may be empty) of the fully constructed paths given
the file ids.
"""
file_objects = self.files(ids, preserve_order)
return [f.make_path(prefix, suffix) for f in file_objects]
[docs] def reverse(self, paths, preserve_order=True):
"""Reverses the lookup from certain paths, returns a list of
:py:class:`bob.db.base.File`'s
Parameters
----------
paths : list
The filename stems (list of str) to query for. This object should be a
python iterable (such as a tuple or list)
preserve_order : :obj:`bool`, optional
If True (the default) the order of elements is preserved, but the
execution time increases.
Returns
-------
list
A list (that may be empty).
"""
file_objects = self.query(self.m_file_class).filter(
self.m_file_class.path.in_(paths))
if not preserve_order:
return file_objects
else:
path_dict = {f.path: f for f in file_objects}
return [path_dict[path] for path in paths]
[docs] def uniquify(self, file_list):
"""Sorts the given list of File objects and removes duplicates from it.
Parameters
----------
file_list : [:py:class:`bob.db.base.File`]
A list of File objects to be handled. Also other objects can be
handled, as long as they are sortable.
Returns
-------
list
A sorted copy of the given ``file_list`` with the duplicates removed.
"""
return sorted(set(file_list))
[docs] def all_files(self, **kwargs):
"""Returns the list of all File objects that satisfy your query.
For possible keyword arguments, please check the implemention's
``objects()`` method.
"""
return self.uniquify(self.objects(**kwargs))
class SQLiteDatabase(SQLiteBaseDatabase, FileDatabase):
"""This class can be used for handling SQL **File** based databases.
It inherits from :py:class:`bob.db.base.SQLiteBaseDatabase` and
:py:class:`bob.db.base.FileDatabase`.
"""
def __init__(self, sqlite_file, file_class,
original_directory, original_extension, **kwargs):
kwargs['sqlite_file'] = sqlite_file
kwargs['file_class'] = file_class
kwargs['original_directory'] = original_directory
kwargs['original_extension'] = original_extension
super(SQLiteDatabase, self).__init__(**kwargs)