helpers.py

#!/usr/bin/env python
# vim: set fileencoding=utf-8 :

###############################################################################
#                                                                             #
# Copyright (c) 2017 Idiap Research Institute, http://www.idiap.ch/           #
# Contact: beat.support@idiap.ch                                              #
#                                                                             #
# This file is part of the beat.backend.python module of the BEAT platform.   #
#                                                                             #
# Commercial License Usage                                                    #
# Licensees holding valid commercial BEAT licenses may use this file in       #
# accordance with the terms contained in a written agreement between you      #
# and Idiap. For further information contact tto@idiap.ch                     #
#                                                                             #
# Alternatively, this file may be used under the terms of the GNU Affero      #
# Public License version 3 as published by the Free Software and appearing    #
# in the file LICENSE.AGPL included in the packaging of this file.            #
# The BEAT platform is distributed in the hope that it will be useful, but    #
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY  #
# or FITNESS FOR A PARTICULAR PURPOSE.                                        #
#                                                                             #
# You should have received a copy of the GNU Affero Public License along      #
# with the BEAT platform. If not, see http://www.gnu.org/licenses/.           #
#                                                                             #
###############################################################################


import os
import errno

import logging
logger = logging.getLogger(__name__)

from . import data
from . import inputs
from . import outputs



def convert_experiment_configuration_to_container(config, proxy_mode):
    data = {
        'proxy_mode': proxy_mode,
      'algorithm': config['algorithm'],
      'parameters': config['parameters'],
      'channel': config['channel'],
      'uid': os.getuid(),
    }

    if 'range' in config:
        data['range'] = config['range']

    data['inputs'] = \
          dict([(k, { 'channel': v['channel'], 'path': v['path'], 'database': v.has_key('database') }) for k,v in config['inputs'].items()])

    if 'outputs' in config:
        data['outputs'] = \
              dict([(k, { 'channel': v['channel'], 'path': v['path'] }) for k,v in config['outputs'].items()])
    else:
        data['result'] = { 'channel': config['channel'], 'path': config['result']['path'] }

    return data


#----------------------------------------------------------


class AccessMode:
    NONE   = 0
    LOCAL  = 1
    REMOTE = 2


def create_inputs_from_configuration(config, algorithm, prefix, cache_root,
                                     cache_access=AccessMode.NONE,
                                     db_access=AccessMode.NONE,
                                     unpack=True, socket=None,
                                     databases=None):

    data_sources = []
    views = {}
    input_list = inputs.InputList()

    # This is used for parallelization purposes
    start_index, end_index = config.get('range', (None, None))

    for name, details in config['inputs'].items():

        if details.get('database', False):
            if db_access == AccessMode.LOCAL:
                if databases is None:
                    raise IOError("No databases provided")

                # Retrieve the database
                try:
                    db = databases[details['database']]
                except:
                    raise IOError("Database '%s' not found" % details['database'])

                # Create of retrieve the database view
                channel = details['channel']

                if not views.has_key(channel):
                    view = db.view(details['protocol'], details['set'])
                    view.prepare_outputs()
                    view.setup()
                    views[channel] = view

                    logger.debug("Database view '%s/%s/%s' created: group='%s'" % \
                                    (details['database'], details['protocol'], details['set'],
                                     channel))
                else:
                    view = views[channel]

                # Creation of the input
                data_source = data.MemoryDataSource(view.done, next_callback=view.next)

                output = view.outputs[details['output']]
                output.data_sink.data_sources.append(data_source)

                input = inputs.Input(name, algorithm.input_map[name], data_source)

                logger.debug("Input '%s' created: group='%s', dataformat='%s', database-output='%s/%s/%s:%s'" % \
                                (name, channel, algorithm.input_map[name], details['database'],
                                 details['protocol'], details['set'], details['output']))

            elif db_access == AccessMode.REMOTE:
                if socket is None:
                    raise IOError("No socket provided for remote inputs")

                input = inputs.RemoteInput(name, algorithm.dataformats[algorithm.input_map[name]],
                                           socket, unpack=unpack)

                logger.debug("RemoteInput '%s' created: group='%s', dataformat='%s', connected to a database" % \
                                (name, details['channel'], algorithm.input_map[name]))

        elif cache_access == AccessMode.LOCAL:
            data_source = data.CachedDataSource()
            data_sources.append(data_source)

            filename = os.path.join(cache_root, details['path'] + '.data')

            if details['channel'] == config['channel']: # synchronized
                status = data_source.setup(
                          filename=filename,
                          prefix=prefix,
                          force_start_index=start_index,
                          force_end_index=end_index,
                          unpack=True,
                         )
            else:
                status = data_source.setup(
                          filename=filename,
                          prefix=prefix,
                          unpack=True,
                         )

            if not status:
                raise IOError("cannot load cache file `%s'" % details['path'])

            input = inputs.Input(name, algorithm.input_map[name], data_source)

            logger.debug("Input '%s' created: group='%s', dataformat='%s', filename='%s'" % \
                            (name, details['channel'], algorithm.input_map[name], filename))

        elif cache_access == AccessMode.REMOTE:
            if socket is None:
                raise IOError("No socket provided for remote inputs")

            input = inputs.RemoteInput(name, algorithm.dataformats[algorithm.input_map[name]],
                                       socket, unpack=unpack)

            logger.debug("RemoteInput '%s' created: group='%s', dataformat='%s'" % \
                            (name, details['channel'], algorithm.input_map[name]))

        else:
            continue

        # Synchronization bits
        group = input_list.group(details['channel'])
        if group is None:
            group = inputs.InputGroup(
                      details['channel'],
                      synchronization_listener=outputs.SynchronizationListener(),
                      restricted_access=(details['channel'] == config['channel'])
                    )
            input_list.add(group)
            logger.debug("Group '%s' created" % details['channel'])

        group.add(input)

    return (input_list, data_sources)



#----------------------------------------------------------


def create_outputs_from_configuration(config, algorithm, prefix, cache_root, input_list,
                                      cache_access=AccessMode.NONE, socket=None):

    data_sinks = []
    output_list = outputs.OutputList()

    # This is used for parallelization purposes
    start_index, end_index = config.get('range', (None, None))

    # If the algorithm is an analyser
    if 'result' in config:
        output_config = {
            'result': config['result']
        }
    else:
        output_config = config['outputs']


    for name, details in output_config.items():

        if 'result' in config:
            dataformat_name = 'analysis:' + algorithm.name
            dataformat = algorithm.result_dataformat()
        else:
            dataformat_name = algorithm.output_map[name]
            dataformat =  algorithm.dataformats[dataformat_name]


        if cache_access == AccessMode.LOCAL:

            path = os.path.join(cache_root, details['path'] + '.data')
            dirname = os.path.dirname(path)
            # Make sure that the directory exists while taking care of race
            # conditions. see: http://stackoverflow.com/questions/273192/check-if-a-directory-exists-and-create-it-if-necessary
            try:
                if (len(dirname) > 0):
                    os.makedirs(dirname)
            except OSError as exception:
                if exception.errno != errno.EEXIST:
                    raise

            data_sink = data.CachedDataSink()
            data_sinks.append(data_sink)

            status = data_sink.setup(
                filename=path,
                dataformat=dataformat,
                encoding='binary',
                max_size=0, # in bytes, for individual file chunks
            )

            if not status:
                raise IOError("Cannot create cache sink '%s'" % details['path'])


            synchronization_listener = None

            if 'result' not in config:
                input_group = input_list.group(details['channel'])
                if (input_group is not None) and hasattr(input_group, 'synchronization_listener'):
                    synchronization_listener = input_group.synchronization_listener

            output_list.add(outputs.Output(name, data_sink,
                synchronization_listener=synchronization_listener,
                force_start_index=start_index or 0)
            )

            if 'result' not in config:
                logger.debug("Output '%s' created: group='%s', dataformat='%s', filename='%s'" % \
                                (name, details['channel'], dataformat_name, path))
            else:
                logger.debug("Output '%s' created: dataformat='%s', filename='%s'" % \
                                (name, dataformat_name, path))

        elif cache_access == AccessMode.REMOTE:
            if socket is None:
                raise IOError("No socket provided for remote outputs")

            output_list.add(outputs.RemoteOutput(name, dataformat, socket))

            logger.debug("RemoteOutput '%s' created: group='%s', dataformat='%s'" % \
                            (name, details['channel'], dataformat_name))

        else:
            continue

    return (output_list, data_sinks)