<commit_before>#!/usr/bin/env python
'''
Provides a data transformation pipeline for expanding the data in
StreamItem instances from streamcorpus.Chunk files.  kba.pipeline.run
provides a command line interface to this functionality.

This software is released under an MIT/X11 open source license.

Copyright 2012 Diffeo, Inc.
'''

import os
import re
import sys
import uuid
#import gevent
import traceback
import itertools
import streamcorpus

from ._logging import log_full_file

## Rather than mess with __import__(), let's just define a mapping
## from strings to the particular things that we expect to see as
## transform functions.  When we want to expose this for user-defined
## transforms, we'll figure out a better interface.
from ._stages import Stages

def _init_stage(name, config):
    '''
    :param name: string name of a stage in Stages

    :param config: config dict passed into the stage constructor

    :returns callable: one of four possible types:

       1) extractors: take byte strings as input and emit StreamItems

       2) incremental transforms: take StreamItem and emit StreamItem
       
       3) batch transforms: take Chunk and emit Chunk

       4) loaders: take Chunk and push it somewhere
    '''
    stage = Stages[name](config)

    ## Note that fromlist must be specified here to cause __import__()
    ## to return the right-most component of name, which in our case
    ## must be a function.  The contents of fromlist is not
    ## considered; it just cannot be empty:
    ## http://stackoverflow.com/questions/2724260/why-does-pythons-import-require-fromlist
    #trans = __import__('clean_html', fromlist=['kba.pipeline'])

    return stage

class Pipeline(object):
    '''    
    configurable pipeline for transforming StreamItem instances in
    streamcorpus.Chunk files into new Chunk files.  Requires a
    configuration dict, which is loaded from a yaml file.
    '''
    def __init__(self, config):
        assert 'kba.pipeline' in config, \
            '"kba.pipeline" missing from config: %r' % config
        config = config['kba.pipeline']
        self.config = config

        ## load the one extractor
        self._extractor = _init_stage(config['extractor'], config)

        ## a list of transforms that take StreamItem instances as
        ## input and emit modified StreamItem instances
        self._incremental_transforms = [
            _init_stage(name, config) 
            for name in config['incremental_transforms']]

        ## a list of transforms that take a chunk path as input and
        ## return a path to a new chunk
        self._batch_transforms = [
            _init_stage(name, config) 
            for name in config['batch_transforms']]

        ## a list of transforms that take a chunk path as input and
        ## return a path to a new chunk
        self._loaders  = [
            _init_stage(name, config) 
            for name in config['loaders']]
<commit_msg>causing the pipeline to handle relative paths to tmp dirs<commit_after>#!/usr/bin/env python
'''
Provides a data transformation pipeline for expanding the data in
StreamItem instances from streamcorpus.Chunk files.  kba.pipeline.run
provides a command line interface to this functionality.

This software is released under an MIT/X11 open source license.

Copyright 2012 Diffeo, Inc.
'''

import os
import re
import sys
import uuid
#import gevent
import traceback
import itertools
import streamcorpus

from ._logging import log_full_file

## Rather than mess with __import__(), let's just define a mapping
## from strings to the particular things that we expect to see as
## transform functions.  When we want to expose this for user-defined
## transforms, we'll figure out a better interface.
from ._stages import Stages

def _init_stage(name, config):
    '''
    :param name: string name of a stage in Stages

    :param config: config dict passed into the stage constructor

    :returns callable: one of four possible types:

       1) extractors: take byte strings as input and emit StreamItems

       2) incremental transforms: take StreamItem and emit StreamItem
       
       3) batch transforms: take Chunk and emit Chunk

       4) loaders: take Chunk and push it somewhere
    '''
    stage = Stages[name](config)

    ## Note that fromlist must be specified here to cause __import__()
    ## to return the right-most component of name, which in our case
    ## must be a function.  The contents of fromlist is not
    ## considered; it just cannot be empty:
    ## http://stackoverflow.com/questions/2724260/why-does-pythons-import-require-fromlist
    #trans = __import__('clean_html', fromlist=['kba.pipeline'])

    return stage

class Pipeline(object):
    '''    
    configurable pipeline for transforming StreamItem instances in
    streamcorpus.Chunk files into new Chunk files.  Requires a
    configuration dict, which is loaded from a yaml file.
    '''
    def __init__(self, config):
        assert 'kba.pipeline' in config, \
            '"kba.pipeline" missing from config: %r' % config
        config = config['kba.pipeline']
        self.config = config

        ## resolve the tmp dir and make sure it exists
        if not config['tmp_dir'].startswith('/'):
            config['tmp_dir'] = os.path.join(os.getcwd(), config['tmp_dir'])

        if not os.path.exists(config['tmp_dir']):
            os.makedirs(config['tmp_dir'])

        ## load the one extractor
        self._extractor = _init_stage(config['extractor'], config)

        ## a list of transforms that take StreamItem instances as
        ## input and emit modified StreamItem instances
        self._incremental_transforms = [
            _init_stage(name, config) 
            for name in config['incremental_transforms']]

        ## a list of transforms that take a chunk path as input and
        ## return a path to a new chunk
        self._batch_transforms = [
            _init_stage(name, config) 
            for name in config['batch_transforms']]

        ## a list of transforms that take a chunk path as input and
        ## return a path to a new chunk
        self._loaders  = [
            _init_stage(name, config) 
            for name in config['loaders']]

{'fraction_non_alphanumeric': 0.07439474106140773, 'fraction_numerical': 0.005451338784672118, 'mean_word_length': 2.9306868304977947, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '98428', 'n_tokens_mistral': 1700, 'n_tokens_neox': 1634, 'n_words': 698}