Commit 21ae26eb authored by André Anjos's avatar André Anjos 💬
Browse files

First implementation

parent 193adcea
#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
# Andre Anjos <andre.anjos@idiap.ch>
# Wed 30 Sep 2015 12:14:50 CEST
import os
from .models import *
from pkg_resources import resource_filename
LOCATION = resource_filename(__name__, 'metadata.csv')
class Database(object):
def __init__(self):
from .driver import Interface
self.info = Interface()
# Loads metadata
import csv
with open(LOCATION) as f:
reader = csv.DictReader(f)
self.metadata = [row for row in reader]
def objects(self):
"""Returns a list of unique :py:class:`.File` objects for the specific
query by the user.
Returns: A list of :py:class:`.File` objects.
"""
return [File(**k) for k in self.metadata]
#!/usr/bin/env python
# encoding: utf-8
# Andre Anjos <andre.anjos@idiap.ch>
# Wed 30 Sep 12:39:50 CEST 2015
"""A script to scan database directories and create a concise CSV descriptor"""
import os
import fnmatch
import bob.io.video
def estimate_duration(video):
"""Estimates the duration of a video clip using Bob"""
v = bob.io.video.reader(video)
return int(v.duration/float(10**6)) #returns in seconds
def scan(args):
"""Scans the given base directory for the database information to retrieve
Yields:
Dictionaries, each with the following fields:
'basedir' (str): The base directory of the sample
'bdf' (str): The stem of the BDF file from its base directory
'video' (str): The stem of the video file from its base directory
'duration' (int): The estimated duration of the video file in seconds
"""
for dirpath, dirs, files in os.walk(args.basedir):
bdf = fnmatch.filter(files, '*.bdf')
video = fnmatch.filter(files, '*C1 trigger*.avi')
if bdf and video: #interesting directory, yield a new dictionary
if args.verbose:
print("Adding BDF file `%s'..." % os.path.join(args.basedir, bdf[0]))
yield {
'basedir': os.path.relpath(dirpath, args.basedir),
'bdf': os.path.splitext(bdf[0])[0],
'video': os.path.splitext(video[0])[0],
'duration': estimate_duration(os.path.join(dirpath, video[0])),
}
def create(args):
"""Creates or re-creates this database"""
from . import LOCATION
if os.path.exists(LOCATION) and not args.recreate:
print("CSV descriptor exists at `%s' and --recreate was not set")
return 1
if os.path.exists(LOCATION): os.unlink(LOCATION)
import csv
with open(LOCATION, 'w') as csvfile:
writer = csv.DictWriter(csvfile, ('basedir','bdf','video','duration'),
delimiter=',')
writer.writeheader()
counter = 0
for row in scan(args):
writer.writerow(row)
counter += 1
if args.verbose:
print("Added %d items to metadata file `%s'" % (counter, LOCATION))
return 0
def add_command(subparsers):
"""Add specific subcommands that the action "create" can use"""
parser = subparsers.add_parser('create', help=create.__doc__)
parser.add_argument('-R', '--recreate', action='store_true', default=False,
help="If set, I'll first erase the current database")
parser.add_argument('-v', '--verbose', action='count', default=0,
help="Do operations in a verbose way")
parser.add_argument('-D', '--basedir', action='store',
default='/idiap/resource/database/HCI_Tagging',
metavar='DIR',
help="Change the relative path to the directory containing the root of the HCI-Tagging database (defaults to %(default)s)")
parser.set_defaults(func=create) #action
#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
# Andre Anjos <andre.anjos@idiap.ch>
# Wed 30 Sep 2015 12:08:31 CEST
"""Commands
"""
import os
import sys
from bob.db.base.driver import Interface as BaseInterface
# Driver API
# ==========
def dumplist(args):
"""Dumps lists of files on the database, inputs your arguments"""
from . import Database
db = Database()
objects = db.objects()
output = sys.stdout
if args.selftest:
from bob.db.base.utils import null
output = null()
for obj in objects:
output.write('%s\n' % (obj.make_path(directory=args.directory, extension=args.extension),))
return 0
def checkfiles(args):
"""Checks the existence of the files based on your criteria"""
from . import Database
db = Database()
objects = db.objects()
# go through all files, check if they are available on the filesystem
good = []
bad = []
for obj in objects:
if os.path.exists(obj.make_path(directory=args.directory, extension=args.extension)): good.append(obj)
else: bad.append(obj)
# report
output = sys.stdout
if args.selftest:
from bob.db.base.utils import null
output = null()
if bad:
for obj in bad:
output.write('Cannot find file "%s"\n' % (obj.make_path(directory=args.directory, extension=args.extension),))
output.write('%d files (out of %d) were not found at "%s"\n' % \
(len(bad), len(objects), args.directory))
return 0
class Interface(BaseInterface):
def name(self):
return 'hci_tagging'
def files(self):
from . import LOCATION
return [LOCATION]
def version(self):
import pkg_resources
return pkg_resources.require('bob.db.%s' % self.name())[0].version
def type(self):
return 'text'
def add_commands(self, parser):
"""Add specific subcommands that the action "dumplist" can use"""
from . import __doc__ as docs
subparsers = self.setup_parser(parser, "Mahnob HCI-Tagging dataset", docs)
# get the "create" action from a submodule
from .create import add_command as create_command
create_command(subparsers)
from argparse import SUPPRESS
# add the dumplist command
dump_message = "Dumps list of files based on your criteria"
dump_parser = subparsers.add_parser('dumplist', help=dump_message)
dump_parser.add_argument('-d', '--directory', dest="directory", default='', help="if given, this path will be prepended to every entry returned (defaults to '%(default)s')")
dump_parser.add_argument('-e', '--extension', dest="extension", default='', help="if given, this extension will be appended to every entry returned (defaults to '%(default)s')")
dump_parser.add_argument('--self-test', dest="selftest", default=False, action='store_true', help=SUPPRESS)
dump_parser.set_defaults(func=dumplist) #action
# add the checkfiles command
check_message = "Check if the files exist, based on your criteria"
check_parser = subparsers.add_parser('checkfiles', help=check_message)
check_parser.add_argument('-d', '--directory', dest="directory", default='', help="if given, this path will be prepended to every entry returned (defaults to '%(default)s')")
check_parser.add_argument('-e', '--extension', dest="extension", default='', help="if given, this extension will be appended to every entry returned (defaults to '%(default)s')")
check_parser.add_argument('--self-test', dest="selftest", default=False, action='store_true', help=SUPPRESS)
check_parser.set_defaults(func=checkfiles) #action
This source diff could not be displayed because it is too large. You can view the blob instead.
#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
# Andre Anjos <andre.anjos@idiap.ch>
# Wed 30 Sep 2015 12:13:47 CEST
import os
import bob
class File(object):
""" Generic file container for HCI-Tagging files
Parameters:
basedir (path): The base directory for the data
bdf (str): The name of the BDF file that accompanies the video file,
containing the physiological signals.
video (str): The name of the video file to be used
duration (int): The time in seconds that corresponds to the estimated
duration of the data (video and physiological signals).
"""
def __init__(self, basedir, bdf, video, duration):
self.basedir = basedir
self.stem = bdf
self.video_stem = video
self.duration = duration
def __repr__(self):
return "File('%s')" % self.stem
def default_extension(self):
return '.bdf'
def make_path(self, directory=None, extension=None):
"""Wraps this files' filename so that a complete path is formed
Keyword parameters:
directory
An optional directory name that will be prefixed to the returned result.
extension
An optional extension that will be suffixed to the returned filename. The
extension normally includes the leading ``.`` character as in ``.png`` or
``.bmp``. If not specified the default extension for the original file in
the database will be used
Returns a string containing the newly generated file path.
"""
return os.path.join(
directory or '',
self.basedir,
self.stem + (extension or self.default_extension()),
)
def save(self, data, directory=None, extension='.hdf5'):
"""Saves the input data at the specified location and using the given
extension.
Keyword parameters:
data
The data blob to be saved (normally a :py:class:`numpy.ndarray`).
directory
If not empty or None, this directory is prefixed to the final file
destination
extension
The extension of the filename - this will control the type of output and
the codec for saving the input blob.
"""
path = self.make_path(directory, extension)
bob.db.utils.makedirs_safe(os.path.dirname(path))
bob.io.save(data, path)
#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
# Andre Anjos <andre.anjos@idiap.ch>
# Fri Mar 14 14:44:31 CET 2014
"""A few checks at the Mahnob HCI-Tagging dataset
"""
import os, sys
import unittest
from . import Database
DATABASE_LOCATION = '/idiap/resource/database/HCI_Tagging'
def db_available(test):
"""Decorator for detecting if we're running the test at Idiap"""
from nose.plugins.skip import SkipTest
import functools
@functools.wraps(test)
def wrapper(*args, **kwargs):
if os.path.exists(DATABASE_LOCATION):
return test(*args, **kwargs)
else:
raise SkipTest("Raw database files are not available")
return wrapper
class HCITaggingTest(unittest.TestCase):
"""Performs various tests on the HCI-Tagging database."""
def setUp(self):
self.db = Database()
def test01_objects(self):
self.assertEqual(len(self.db.objects()), 3490)
class CmdLineTest(unittest.TestCase):
"""Makes sure our command-line is working properly."""
def test01_manage_dumplist(self):
from bob.db.base.script.dbmanage import main
args = [
'hci_tagging',
'dumplist',
'--self-test',
]
self.assertEqual(main(args), 0)
@db_available
def test02_manage_checkfiles(self):
from bob.db.base.script.dbmanage import main
args = [
'hci_tagging',
'checkfiles',
'--self-test',
'--directory=%s' % DATABASE_LOCATION,
]
self.assertEqual(main(args), 0)
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