streamcorpus_pipeline/_s3_storage.py
def _decode(self, data):
        '''
        Given the raw data from s3, return a generator for the items
        contained in that data. A generator is necessary to support
        chunk files, but non-chunk files can be provided by a generator
        that yields exactly one item.

        Decoding works by case analysis on the config option
        ``input_format``. If an invalid ``input_format`` is given, then
        a ``ConfigurationError`` is raised.
        '''
        informat = self.config['input_format'].lower()
        if informat == 'spinn3r':
            return _generate_stream_items(data)
        elif informat == 'streamitem':
            ver = self.config['streamcorpus_version']
            if ver not in _message_versions:
                raise ConfigurationError(
                    'Not a valid streamcorpus version: %s '
                    '(choose from: %s)'
                    % (ver, ', '.join(_message_versions.keys())))

            message = _message_versions[ver]
            return streamcorpus.Chunk(data=data, message=message)
        elif informat == 'featurecollection' and FCChunk is not None:
            return FCChunk(data=data)
        else:
            raise ConfigurationError(
                'from_s3_chunks unknown input_format = %r'
                % informat)
def _decode(self, data):
        '''
        Given the raw data from s3, return a generator for the items
        contained in that data. A generator is necessary to support
        chunk files, but non-chunk files can be provided by a generator
        that yields exactly one item.

        Decoding works by case analysis on the config option
        ``input_format``. If an invalid ``input_format`` is given, then
        a ``ConfigurationError`` is raised.
        '''
        informat = self.config['input_format'].lower()
        if informat == 'spinn3r':
            return _generate_stream_items(data)
        elif informat == 'streamitem':
            ver = self.config['streamcorpus_version']
            if ver not in _message_versions:
                raise ConfigurationError(
                    'Not a valid streamcorpus version: %s '
                    '(choose from: %s)'
                    % (ver, ', '.join(_message_versions.keys())))

            message = _message_versions[ver]
            return streamcorpus.Chunk(data=data, message=message)
        elif informat == 'featurecollection' and FCChunk is not None:
            return FCChunk(data=data)
        else:
            raise ConfigurationError(
                'from_s3_chunks unknown input_format = %r'
                % informat)
Given the raw data from s3, return a generator for the items
        contained in that data. A generator is necessary to support
        chunk files, but non-chunk files can be provided by a generator
        that yields exactly one item.

        Decoding works by case analysis on the config option
        ``input_format``. If an invalid ``input_format`` is given, then
        a ``ConfigurationError`` is raised.